ligo.at
decentralized and customizable links page on top of atproto
ligo.at
atproto
link-in-bio
python
uv
add link sections
nauta.one
+50
-25
src/main.py
+50
-25
src/main.py
···
1
import asyncio
2
import json
3
4
-
from aiohttp.client import ClientResponse, ClientSession
5
-
from flask import Flask, g, session, redirect, render_template, request, url_for
6
-
from flask_htmx import HTMX, make_response as htmx_reponse
7
-
from typing import Any
8
9
from .atproto import (
10
PdsUrl,
···
25
htmx = HTMX()
26
htmx.init_app(app)
27
init_db(app)
28
-
29
-
SCHEMA = "at.ligo"
30
31
32
@app.before_request
···
74
pds = await resolve_pds_from_did(client, did=did, kv=pdskv, reload=reload)
75
if pds is None:
76
return render_template("error.html", message="pds not found"), 404
77
-
(profile, _), links = await asyncio.gather(
78
load_profile(client, pds, did, reload=reload),
79
load_links(client, pds, did, reload=reload),
80
)
81
-
if links is None:
82
return render_template("error.html", message="profile not found"), 404
83
84
if reload:
85
# remove the ?reload parameter
86
return redirect(request.path)
87
88
-
profile["handle"] = handle
89
athref = f"at://{did}/at.ligo.actor.links/self"
90
return render_template(
91
"profile.html",
92
profile=profile,
93
-
links=links,
94
athref=athref,
95
)
96
···
129
handle: str | None = user.handle
130
131
async with ClientSession() as client:
132
-
(profile, from_bluesky), links = await asyncio.gather(
133
load_profile(client, pds, did),
134
load_links(client, pds, did),
135
)
136
137
return render_template(
138
"editor.html",
139
handle=handle,
140
profile=profile,
141
profile_from_bluesky=from_bluesky,
142
-
links=json.dumps(links or []),
143
)
144
145
···
155
return redirect("/editor", 303)
156
157
record = {
158
-
"$type": f"{SCHEMA}.actor.profile",
159
"displayName": display_name,
160
"description": description,
161
}
···
164
user=user,
165
pds=user.pds_url,
166
repo=user.did,
167
-
collection=f"{SCHEMA}.actor.profile",
168
rkey="self",
169
record=record,
170
)
···
174
kv.set(user.did, json.dumps(record))
175
176
if htmx:
177
-
return htmx_reponse(
178
render_template("_editor_profile.html", profile=record),
179
reswap="outerHTML",
180
)
···
206
links.append(link)
207
208
record = {
209
-
"$type": f"{SCHEMA}.actor.links",
210
-
"links": links,
211
}
212
213
success = await put_record(
214
user=user,
215
pds=user.pds_url,
216
repo=user.did,
217
-
collection=f"{SCHEMA}.actor.links",
218
rkey="self",
219
record=record,
220
)
···
224
kv.set(user.did, json.dumps(record))
225
226
if htmx:
227
-
return htmx_reponse(
228
-
render_template("_editor_links.html", links=record["links"]),
229
reswap="outerHTML",
230
)
231
···
237
return render_template("terms.html")
238
239
240
async def load_links(
241
client: ClientSession,
242
pds: str,
243
did: str,
244
reload: bool = False,
245
-
) -> list[dict[str, str]] | None:
246
kv = KV(app, app.logger, "links_from_did")
247
record_json = kv.get(did)
248
249
if record_json is not None and not reload:
250
-
return json.loads(record_json)["links"]
251
252
-
record = await get_record(client, pds, did, f"{SCHEMA}.actor.links", "self")
253
if record is None:
254
return None
255
256
kv.set(did, value=json.dumps(record))
257
-
return record["links"]
258
259
260
async def load_profile(
···
271
return json.loads(record_json), False
272
273
(record, bsky_record) = await asyncio.gather(
274
-
get_record(client, pds, did, f"{SCHEMA}.actor.profile", "self"),
275
get_record(client, pds, did, "app.bsky.actor.profile", "self"),
276
)
277
···
1
import asyncio
2
import json
3
+
from typing import Any, NamedTuple
4
5
+
from aiohttp.client import ClientSession
6
+
from flask import Flask, g, redirect, render_template, request, session, url_for
7
+
from flask_htmx import HTMX
8
+
from flask_htmx import make_response as htmx_response
9
10
from .atproto import (
11
PdsUrl,
···
26
htmx = HTMX()
27
htmx.init_app(app)
28
init_db(app)
29
30
31
@app.before_request
···
73
pds = await resolve_pds_from_did(client, did=did, kv=pdskv, reload=reload)
74
if pds is None:
75
return render_template("error.html", message="pds not found"), 404
76
+
(profile, _), link_sections = await asyncio.gather(
77
load_profile(client, pds, did, reload=reload),
78
load_links(client, pds, did, reload=reload),
79
)
80
+
if profile is None or link_sections is None:
81
return render_template("error.html", message="profile not found"), 404
82
83
if reload:
84
# remove the ?reload parameter
85
return redirect(request.path)
86
87
+
if handle:
88
+
profile["handle"] = handle
89
athref = f"at://{did}/at.ligo.actor.links/self"
90
return render_template(
91
"profile.html",
92
profile=profile,
93
+
links=link_sections[0].links,
94
+
sections=link_sections,
95
athref=athref,
96
)
97
···
130
handle: str | None = user.handle
131
132
async with ClientSession() as client:
133
+
(profile, from_bluesky), link_sections = await asyncio.gather(
134
load_profile(client, pds, did),
135
load_links(client, pds, did),
136
)
137
138
+
links = []
139
+
if link_sections:
140
+
links = link_sections[0].links
141
+
142
return render_template(
143
"editor.html",
144
handle=handle,
145
profile=profile,
146
profile_from_bluesky=from_bluesky,
147
+
links=json.dumps(links),
148
)
149
150
···
160
return redirect("/editor", 303)
161
162
record = {
163
+
"$type": "at.ligo.actor.profile",
164
"displayName": display_name,
165
"description": description,
166
}
···
169
user=user,
170
pds=user.pds_url,
171
repo=user.did,
172
+
collection="at.ligo.actor.profile",
173
rkey="self",
174
record=record,
175
)
···
179
kv.set(user.did, json.dumps(record))
180
181
if htmx:
182
+
return htmx_response(
183
render_template("_editor_profile.html", profile=record),
184
reswap="outerHTML",
185
)
···
211
links.append(link)
212
213
record = {
214
+
"$type": "at.ligo.actor.links",
215
+
"sections": [
216
+
{
217
+
"title": "",
218
+
"links": links,
219
+
}
220
+
],
221
}
222
223
success = await put_record(
224
user=user,
225
pds=user.pds_url,
226
repo=user.did,
227
+
collection="at.ligo.actor.links",
228
rkey="self",
229
record=record,
230
)
···
234
kv.set(user.did, json.dumps(record))
235
236
if htmx:
237
+
return htmx_response(
238
+
render_template("_editor_links.html", links=record["sections"][0]["links"]),
239
reswap="outerHTML",
240
)
241
···
247
return render_template("terms.html")
248
249
250
+
class LinkSection(NamedTuple):
251
+
title: str
252
+
links: list[dict[str, str]]
253
+
254
+
255
async def load_links(
256
client: ClientSession,
257
pds: str,
258
did: str,
259
reload: bool = False,
260
+
) -> list[LinkSection] | None:
261
kv = KV(app, app.logger, "links_from_did")
262
record_json = kv.get(did)
263
264
if record_json is not None and not reload:
265
+
parsed = json.loads(record_json)
266
+
return _links_or_sections(parsed)
267
268
+
record = await get_record(client, pds, did, "at.ligo.actor.links", "self")
269
if record is None:
270
return None
271
272
kv.set(did, value=json.dumps(record))
273
+
return _links_or_sections(record)
274
+
275
+
276
+
def _links_or_sections(raw: dict[str, Any]) -> list[LinkSection] | None:
277
+
if "sections" in raw:
278
+
return list(map(lambda s: LinkSection(**s), raw["sections"]))
279
+
elif "links" in raw:
280
+
return [LinkSection("", raw["links"])]
281
+
else:
282
+
return None
283
284
285
async def load_profile(
···
296
return json.loads(record_json), False
297
298
(record, bsky_record) = await asyncio.gather(
299
+
get_record(client, pds, did, "at.ligo.actor.profile", "self"),
300
get_record(client, pds, did, "app.bsky.actor.profile", "self"),
301
)
302
+11
-11
src/oauth.py
+11
-11
src/oauth.py
···
1
from aiohttp.client import ClientSession
2
from authlib.jose import JsonWebKey, Key
3
from flask import Blueprint, current_app, jsonify, redirect, request, session, url_for
4
-
from urllib.parse import urlencode
5
-
6
-
import json
7
8
-
from .auth import (
9
-
delete_auth_request,
10
-
get_auth_request,
11
-
save_auth_request,
12
-
save_auth_session,
13
-
)
14
-
from .db import KV, get_db
15
from .atproto import (
16
is_valid_did,
17
is_valid_handle,
18
pds_endpoint_from_doc,
19
resolve_authserver_from_pds,
20
-
fetch_authserver_meta,
21
resolve_identity,
22
)
23
from .atproto.oauth import initial_token_request, send_par_auth_request
24
from .atproto.types import OAuthAuthRequest, OAuthSession
25
from .security import is_safe_url
26
27
oauth = Blueprint("oauth", __name__, url_prefix="/oauth")
···
1
+
import json
2
+
from urllib.parse import urlencode
3
+
4
from aiohttp.client import ClientSession
5
from authlib.jose import JsonWebKey, Key
6
from flask import Blueprint, current_app, jsonify, redirect, request, session, url_for
7
8
from .atproto import (
9
+
fetch_authserver_meta,
10
is_valid_did,
11
is_valid_handle,
12
pds_endpoint_from_doc,
13
resolve_authserver_from_pds,
14
resolve_identity,
15
)
16
from .atproto.oauth import initial_token_request, send_par_auth_request
17
from .atproto.types import OAuthAuthRequest, OAuthSession
18
+
from .auth import (
19
+
delete_auth_request,
20
+
get_auth_request,
21
+
save_auth_request,
22
+
save_auth_session,
23
+
)
24
+
from .db import KV, get_db
25
from .security import is_safe_url
26
27
oauth = Blueprint("oauth", __name__, url_prefix="/oauth")
+2
-1
src/static/profile/default.css
+2
-1
src/static/profile/default.css
+4
src/static/style.css
+4
src/static/style.css
+4
-2
src/templates/profile.html
+4
-2
src/templates/profile.html
···
35
{% endif %}
36
</header>
37
<div class="links-container">
38
-
<ul>
39
-
{% for link in links %}
40
<li style="color: {{ link.backgroundColor }}">
41
<a href="{{ link.href }}">
42
<div class="link-item">
···
49
</li>
50
{% endfor %}
51
</ul>
52
</div>
53
<footer>
54
made with <a href="https://ligo.at" target="_self">ligo.at</a>
···
35
{% endif %}
36
</header>
37
<div class="links-container">
38
+
{% for section in sections %}
39
+
<ul class="link-section">
40
+
{% for link in section.links %}
41
<li style="color: {{ link.backgroundColor }}">
42
<a href="{{ link.href }}">
43
<div class="link-item">
···
50
</li>
51
{% endfor %}
52
</ul>
53
+
{% endfor %}
54
</div>
55
<footer>
56
made with <a href="https://ligo.at" target="_self">ligo.at</a>