Skip to content

Storage

Your database, from socialchimp's point of view: five methods, and nothing about how or where the rows are kept. See why the library never touches your database for the reasoning, and getting started for a worked example.

Storage

Storage

Bases: Protocol

What your app must provide so socialchimp can save things.

Five methods. Every one is async. If the database layer you have is a blocking one, write the five methods the ordinary way and hand the class to sync_storage below, which does the wrapping for you. Django has its own version of that, contrib.django.orm_storage, because its ORM cares which thread it is run on.

None of these should raise when something is missing. Return None instead, and let deleting something that is not there pass quietly.

get_connection async 

get_connection(connection_id: str) -> Connection | None

Look up one connected account.

Parameters:

Name Type Description Default
connection_id str

The id your app gave this connection.

 required

Returns:

Type Description
Connection | None

The connection, or None if there is no such connection.
Source code in src/socialchimp/storage.py 
57
58
59
60
61
62
63
64
65
66
async def get_connection(self, connection_id: str) -> Connection | None:
    """Look up one connected account.

    Args:
        connection_id: The id your app gave this connection.

    Returns:
        The connection, or `None` if there is no such connection.
    """
    ...

save_connection async 

save_connection(connection: Connection) -> None

Write a connection, replacing any earlier one with the same id.

Called when an account is first connected, and again every time a token is renewed.

Parameters:

Name Type Description Default
connection Connection

The connection to write.

 required
Source code in src/socialchimp/storage.py 
68
69
70
71
72
73
74
75
76
77
async def save_connection(self, connection: Connection) -> None:
    """Write a connection, replacing any earlier one with the same id.

    Called when an account is first connected, and again every time a
    token is renewed.

    Args:
        connection: The connection to write.
    """
    ...

delete_connection async 

delete_connection(connection_id: str) -> None

Remove a connection. Quiet if it is already gone.

Parameters:

Name Type Description Default
connection_id str

The id your app gave this connection.

 required
Source code in src/socialchimp/storage.py 
79
80
81
82
83
84
85
async def delete_connection(self, connection_id: str) -> None:
    """Remove a connection. Quiet if it is already gone.

    Args:
        connection_id: The id your app gave this connection.
    """
    ...

get_app async 

get_app(
    platform: str, host: str | None
) -> AppCredentials | None

Look up your app's credentials for one network.

Parameters:

Name Type Description Default
platform str

Which network, for example "mastodon".

 required
host str | None

Which server, for networks that have more than one. None for networks with a single server.

 required

Returns:

Type Description
AppCredentials | None

The credentials, or None if none are stored yet.
Source code in src/socialchimp/storage.py 
87
88
89
90
91
92
93
94
95
96
97
98
async def get_app(self, platform: str, host: str | None) -> AppCredentials | None:
    """Look up your app's credentials for one network.

    Args:
        platform: Which network, for example `"mastodon"`.
        host: Which server, for networks that have more than one.
            `None` for networks with a single server.

    Returns:
        The credentials, or `None` if none are stored yet.
    """
    ...

save_app async 

save_app(app: AppCredentials) -> None

Write your app's credentials for one network.

Mostly used after socialchimp registers an app on a Mastodon server for you, since that has to happen once per server.

Parameters:

Name Type Description Default
app AppCredentials

The credentials to write.

 required
Source code in src/socialchimp/storage.py 
100
101
102
103
104
105
106
107
108
109
async def save_app(self, app: AppCredentials) -> None:
    """Write your app's credentials for one network.

    Mostly used after socialchimp registers an app on a Mastodon server
    for you, since that has to happen once per server.

    Args:
        app: The credentials to write.
    """
    ...

For a blocking database layer

Write the five methods the ordinary, blocking way - the Django ORM, a psycopg cursor, a SQLAlchemy session - and hand the class to sync_storage. Every call then runs on a spare thread.

SyncStorage

Bases: Protocol

Storage, written the ordinary blocking way.

The same five methods, none of them async. Most apps with a database already have a layer like this - the Django ORM, a psycopg cursor, a SQLAlchemy session - and there is no reason to rewrite it as async code just to keep socialchimp happy.

Hand one to sync_storage and it becomes a Storage the core can use.

get_connection 

get_connection(connection_id: str) -> Connection | None

Look up one connected account.

Parameters:

Name Type Description Default
connection_id str

The id your app gave this connection.

 required

Returns:

Type Description
Connection | None

The connection, or None if there is no such connection.
Source code in src/socialchimp/storage.py 
184
185
186
187
188
189
190
191
192
193
def get_connection(self, connection_id: str) -> Connection | None:
    """Look up one connected account.

    Args:
        connection_id: The id your app gave this connection.

    Returns:
        The connection, or `None` if there is no such connection.
    """
    ...

save_connection 

save_connection(connection: Connection) -> None

Write a connection, replacing any earlier one with the same id.

Parameters:

Name Type Description Default
connection Connection

The connection to write.

 required
Source code in src/socialchimp/storage.py 
195
196
197
198
199
200
201
def save_connection(self, connection: Connection) -> None:
    """Write a connection, replacing any earlier one with the same id.

    Args:
        connection: The connection to write.
    """
    ...

delete_connection 

delete_connection(connection_id: str) -> None

Remove a connection. Quiet if it is already gone.

Parameters:

Name Type Description Default
connection_id str

The id your app gave this connection.

 required
Source code in src/socialchimp/storage.py 
203
204
205
206
207
208
209
def delete_connection(self, connection_id: str) -> None:
    """Remove a connection. Quiet if it is already gone.

    Args:
        connection_id: The id your app gave this connection.
    """
    ...

get_app 

get_app(
    platform: str, host: str | None
) -> AppCredentials | None

Look up your app's credentials for one network.

Parameters:

Name Type Description Default
platform str

Which network.

 required
host str | None

Which server, or None.

 required

Returns:

Type Description
AppCredentials | None

The credentials, or None if none are stored yet.
Source code in src/socialchimp/storage.py 
211
212
213
214
215
216
217
218
219
220
221
def get_app(self, platform: str, host: str | None) -> AppCredentials | None:
    """Look up your app's credentials for one network.

    Args:
        platform: Which network.
        host: Which server, or `None`.

    Returns:
        The credentials, or `None` if none are stored yet.
    """
    ...

save_app 

save_app(app: AppCredentials) -> None

Write your app's credentials for one network.

Parameters:

Name Type Description Default
app AppCredentials

The credentials to write.

 required
Source code in src/socialchimp/storage.py 
223
224
225
226
227
228
229
def save_app(self, app: AppCredentials) -> None:
    """Write your app's credentials for one network.

    Args:
        app: The credentials to write.
    """
    ...

sync_storage 

sync_storage(
    inner: SyncStorage, *, run: RunInThread | None = None
) -> Storage

Let the core use a storage class you wrote as blocking code.

Example

class MyStorage: def get_connection(self, connection_id): row = session.get(SocialAccount, connection_id) return row.to_connection() if row else None ...

sc = SocialChimp(storage=sync_storage(MyStorage()))

Parameters:

Name Type Description Default
inner SyncStorage

Your storage class, with the five methods written the ordinary way.

 required
run RunInThread | None

How to run one of those methods. Left out, each call goes to a spare thread, which is right for anything but Django - see socialchimp.contrib.django.orm_storage.

 None

Returns:

Type Description
Storage

A Storage to hand to SocialChimp.
Source code in src/socialchimp/storage.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
def sync_storage(inner: SyncStorage, *, run: RunInThread | None = None) -> Storage:
    """Let the core use a storage class you wrote as blocking code.

    Example:
        class MyStorage:
            def get_connection(self, connection_id):
                row = session.get(SocialAccount, connection_id)
                return row.to_connection() if row else None
            ...

        sc = SocialChimp(storage=sync_storage(MyStorage()))

    Args:
        inner: Your storage class, with the five methods written the
            ordinary way.
        run: How to run one of those methods. Left out, each call goes to a
            spare thread, which is right for anything but Django - see
            `socialchimp.contrib.django.orm_storage`.

    Returns:
        A `Storage` to hand to `SocialChimp`.
    """
    return _StorageInAThread(inner, run if run is not None else in_a_thread)

RunInThread

Bases: Protocol

Runs one piece of blocking work without blocking the event loop.

There is more than one right answer to this, which is why it is a setting rather than a decision. in_a_thread hands the work to any spare thread, which is what a plain app wants. Django wants it run on the thread the request arrived on, because that is where its database connection lives - see socialchimp.contrib.django.

in_a_thread async 

in_a_thread(work: Callable[[], T]) -> T

Run blocking work on a spare thread.

Parameters:

Name Type Description Default
work Callable[[], T]

The blocking call, already given its arguments.

 required

Returns:

Type Description
T

Whatever the work returned.
Source code in src/socialchimp/storage.py 
254
255
256
257
258
259
260
261
262
263
async def in_a_thread(work: Callable[[], T]) -> T:
    """Run blocking work on a spare thread.

    Args:
        work: The blocking call, already given its arguments.

    Returns:
        Whatever the work returned.
    """
    return await asyncio.to_thread(work)

Trying things out

Forgets everything when your program stops. Fine for a first look; not for production.

InMemoryStorage 

InMemoryStorage()

Storage that keeps everything in memory and forgets it on restart.

Useful for tests, examples and a first look at the library. Do not use it in production - every restart disconnects every account.

Start with nothing stored.

Source code in src/socialchimp/storage.py 
119
120
121
122
def __init__(self) -> None:
    """Start with nothing stored."""
    self._connections: dict[str, Connection] = {}
    self._apps: dict[tuple[str, str | None], AppCredentials] = {}

get_connection async 

get_connection(connection_id: str) -> Connection | None

Look up one connected account.

Parameters:

Name Type Description Default
connection_id str

The id your app gave this connection.

 required

Returns:

Type Description
Connection | None

The connection, or None if there is no such connection.
Source code in src/socialchimp/storage.py 
124
125
126
127
128
129
130
131
132
133
async def get_connection(self, connection_id: str) -> Connection | None:
    """Look up one connected account.

    Args:
        connection_id: The id your app gave this connection.

    Returns:
        The connection, or `None` if there is no such connection.
    """
    return self._connections.get(connection_id)

save_connection async 

save_connection(connection: Connection) -> None

Write a connection, replacing any earlier one with the same id.

Parameters:

Name Type Description Default
connection Connection

The connection to write.

 required
Source code in src/socialchimp/storage.py 
135
136
137
138
139
140
141
async def save_connection(self, connection: Connection) -> None:
    """Write a connection, replacing any earlier one with the same id.

    Args:
        connection: The connection to write.
    """
    self._connections[connection.id] = connection

delete_connection async 

delete_connection(connection_id: str) -> None

Remove a connection. Quiet if it is already gone.

Parameters:

Name Type Description Default
connection_id str

The id your app gave this connection.

 required
Source code in src/socialchimp/storage.py 
143
144
145
146
147
148
149
async def delete_connection(self, connection_id: str) -> None:
    """Remove a connection. Quiet if it is already gone.

    Args:
        connection_id: The id your app gave this connection.
    """
    self._connections.pop(connection_id, None)

get_app async 

get_app(
    platform: str, host: str | None
) -> AppCredentials | None

Look up your app's credentials for one network.

Parameters:

Name Type Description Default
platform str

Which network.

 required
host str | None

Which server, or None.

 required

Returns:

Type Description
AppCredentials | None

The credentials, or None if none are stored yet.
Source code in src/socialchimp/storage.py 
151
152
153
154
155
156
157
158
159
160
161
async def get_app(self, platform: str, host: str | None) -> AppCredentials | None:
    """Look up your app's credentials for one network.

    Args:
        platform: Which network.
        host: Which server, or `None`.

    Returns:
        The credentials, or `None` if none are stored yet.
    """
    return self._apps.get((platform, host))

save_app async 

save_app(app: AppCredentials) -> None

Write your app's credentials for one network.

Parameters:

Name Type Description Default
app AppCredentials

The credentials to write.

 required
Source code in src/socialchimp/storage.py 
163
164
165
166
167
168
169
async def save_app(self, app: AppCredentials) -> None:
    """Write your app's credentials for one network.

    Args:
        app: The credentials to write.
    """
    self._apps[app.key] = app