decentralized and customizable links page on top of atproto
ligo.at
atproto
link-in-bio
python
uv
cache profile and links when updating them
+48
-30
+48
-30
src/main.py
+48
-30
src/main.py
···
120
120
121
121
async with ClientSession() as client:
122
122
(profile, from_bluesky), links = await asyncio.gather(
123
-
load_profile(client, pds, did, reload=True),
124
-
load_links(client, pds, did, reload=True),
123
+
load_profile(client, pds, did),
124
+
load_links(client, pds, did),
125
125
)
126
126
127
127
return render_template(
···
140
140
return redirect("/login", 303)
141
141
142
142
display_name = request.form.get("displayName")
143
-
description = request.form.get("description") or ""
143
+
description = request.form.get("description", "")
144
144
if not display_name:
145
145
return redirect("/editor", 303)
146
146
147
-
await put_record(
147
+
record = {
148
+
"$type": f"{SCHEMA}.actor.profile",
149
+
"displayName": display_name,
150
+
"description": description,
151
+
}
152
+
153
+
success = await put_record(
148
154
user=user,
149
155
pds=user.pds_url,
150
156
repo=user.did,
151
157
collection=f"{SCHEMA}.actor.profile",
152
158
rkey="self",
153
-
record={
154
-
"$type": f"{SCHEMA}.actor.profile",
155
-
"displayName": display_name,
156
-
"description": description,
157
-
},
159
+
record=record,
158
160
)
159
161
160
-
return redirect("/editor", 303)
162
+
if success:
163
+
kv = KV(app, app.logger, "profile_from_did")
164
+
kv.set(user.did, json.dumps(record))
165
+
166
+
return redirect(url_for("page_editor"), 303)
161
167
162
168
163
169
@app.post("/editor/links")
···
183
189
link["subtitle"] = subtitle
184
190
links.append(link)
185
191
186
-
await put_record(
192
+
record = {
193
+
"$type": f"{SCHEMA}.actor.links",
194
+
"links": links,
195
+
}
196
+
197
+
success = await put_record(
187
198
user=user,
188
199
pds=user.pds_url,
189
200
repo=user.did,
190
201
collection=f"{SCHEMA}.actor.links",
191
202
rkey="self",
192
-
record={
193
-
"$type": f"{SCHEMA}.actor.links",
194
-
"links": links,
195
-
},
203
+
record=record,
196
204
)
197
205
198
-
return redirect("/editor", 303)
206
+
if success:
207
+
kv = KV(app, app.logger, "links_from_did")
208
+
kv.set(user.did, json.dumps(record))
209
+
210
+
return redirect(url_for("page_editor"), 303)
199
211
200
212
201
213
@app.get("/terms")
···
210
222
reload: bool = False,
211
223
) -> list[dict[str, str]] | None:
212
224
kv = KV(app, app.logger, "links_from_did")
213
-
recordstr = kv.get(did)
225
+
record_json = kv.get(did)
214
226
215
-
if recordstr is not None and not reload:
216
-
return json.loads(recordstr)["links"]
227
+
if record_json is not None and not reload:
228
+
return json.loads(record_json)["links"]
217
229
218
230
record = await get_record(client, pds, did, f"{SCHEMA}.actor.links", "self")
219
231
if record is None:
···
231
243
reload: bool = False,
232
244
) -> tuple[dict[str, str] | None, bool]:
233
245
kv = KV(app, app.logger, "profile_from_did")
234
-
recordstr = kv.get(did)
246
+
record_json = kv.get(did)
247
+
248
+
if record_json is not None and not reload:
249
+
return json.loads(record_json), False
235
250
236
-
if recordstr is not None and not reload:
237
-
return json.loads(recordstr), False
251
+
(record, bsky_record) = await asyncio.gather(
252
+
get_record(client, pds, did, f"{SCHEMA}.actor.profile", "self"),
253
+
get_record(client, pds, did, "app.bsky.actor.profile", "self"),
254
+
)
238
255
239
256
from_bluesky = False
240
-
record = await get_record(client, pds, did, f"{SCHEMA}.actor.profile", "self")
241
257
if record is None and fallback_with_bluesky:
242
-
record = await get_record(client, pds, did, "app.bsky.actor.profile", "self")
258
+
record = bsky_record
243
259
from_bluesky = True
244
-
if record is None:
245
-
return None, False
260
+
261
+
if record is not None:
262
+
kv.set(did, value=json.dumps(record))
246
263
247
-
kv.set(did, value=json.dumps(record))
248
264
return record, from_bluesky
249
265
250
266
···
256
272
collection: str,
257
273
rkey: str,
258
274
record: dict[str, Any],
259
-
):
275
+
) -> bool:
276
+
"""Writes the record onto the users PDS. Returns bool for success."""
277
+
260
278
endpoint = f"{pds}/xrpc/com.atproto.repo.putRecord"
261
279
body = {
262
280
"repo": repo,
···
276
294
user=user,
277
295
update_dpop_pds_nonce=update_dpop_pds_nonce,
278
296
)
279
-
if not response or not response.ok:
280
-
app.logger.warning("PDS HTTP ERROR")
297
+
298
+
return response is not None and response.ok
281
299
282
300
283
301
def _is_did_blocked(did: str) -> bool: