mirror of https://git.sr.ht/~tsileo/microblog.pub
Compare commits
2 Commits
c506299089
...
673baf0d7f
Author | SHA1 | Date |
---|---|---|
Thomas Sileo | 673baf0d7f | |
Thomas Sileo | 9c65919070 |
21
app/main.py
21
app/main.py
|
@ -1640,23 +1640,26 @@ async def json_feed(
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
result = {
|
result = {
|
||||||
"version": "https://jsonfeed.org/version/1",
|
"version": "https://jsonfeed.org/version/1.1",
|
||||||
"title": f"{LOCAL_ACTOR.display_name}'s microblog'",
|
"title": f"{LOCAL_ACTOR.display_name}'s microblog'",
|
||||||
"home_page_url": LOCAL_ACTOR.url,
|
"home_page_url": LOCAL_ACTOR.url,
|
||||||
"feed_url": BASE_URL + "/feed.json",
|
"feed_url": BASE_URL + "/feed.json",
|
||||||
"author": {
|
"authors": [
|
||||||
|
{
|
||||||
"name": LOCAL_ACTOR.display_name,
|
"name": LOCAL_ACTOR.display_name,
|
||||||
"url": LOCAL_ACTOR.url,
|
"url": LOCAL_ACTOR.url,
|
||||||
},
|
}
|
||||||
|
],
|
||||||
"items": data,
|
"items": data,
|
||||||
}
|
}
|
||||||
if LOCAL_ACTOR.icon_url:
|
if LOCAL_ACTOR.icon_url:
|
||||||
result["author"]["avatar"] = LOCAL_ACTOR.icon_url # type: ignore
|
result["authors"][0]["avatar"] = LOCAL_ACTOR.icon_url # type: ignore
|
||||||
return result
|
return result
|
||||||
|
|
||||||
|
|
||||||
async def _gen_rss_feed(
|
async def _gen_rss_feed(
|
||||||
db_session: AsyncSession,
|
db_session: AsyncSession,
|
||||||
|
is_rss: bool,
|
||||||
):
|
):
|
||||||
fg = FeedGenerator()
|
fg = FeedGenerator()
|
||||||
fg.id(BASE_URL + "/feed.rss")
|
fg.id(BASE_URL + "/feed.rss")
|
||||||
|
@ -1687,8 +1690,12 @@ async def _gen_rss_feed(
|
||||||
|
|
||||||
fe = fg.add_entry()
|
fe = fg.add_entry()
|
||||||
fe.id(outbox_object.url)
|
fe.id(outbox_object.url)
|
||||||
fe.link(href=outbox_object.url)
|
|
||||||
|
# Atom feeds require a title
|
||||||
|
if not is_rss:
|
||||||
fe.title(outbox_object.url)
|
fe.title(outbox_object.url)
|
||||||
|
|
||||||
|
fe.link(href=outbox_object.url)
|
||||||
fe.description(content)
|
fe.description(content)
|
||||||
fe.content(content)
|
fe.content(content)
|
||||||
fe.published(outbox_object.ap_published_at.replace(tzinfo=timezone.utc))
|
fe.published(outbox_object.ap_published_at.replace(tzinfo=timezone.utc))
|
||||||
|
@ -1701,7 +1708,7 @@ async def rss_feed(
|
||||||
db_session: AsyncSession = Depends(get_db_session),
|
db_session: AsyncSession = Depends(get_db_session),
|
||||||
) -> PlainTextResponse:
|
) -> PlainTextResponse:
|
||||||
return PlainTextResponse(
|
return PlainTextResponse(
|
||||||
(await _gen_rss_feed(db_session)).rss_str(),
|
(await _gen_rss_feed(db_session, is_rss=True)).rss_str(),
|
||||||
headers={"Content-Type": "application/rss+xml"},
|
headers={"Content-Type": "application/rss+xml"},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -1711,6 +1718,6 @@ async def atom_feed(
|
||||||
db_session: AsyncSession = Depends(get_db_session),
|
db_session: AsyncSession = Depends(get_db_session),
|
||||||
) -> PlainTextResponse:
|
) -> PlainTextResponse:
|
||||||
return PlainTextResponse(
|
return PlainTextResponse(
|
||||||
(await _gen_rss_feed(db_session)).atom_str(),
|
(await _gen_rss_feed(db_session, is_rss=False)).atom_str(),
|
||||||
headers={"Content-Type": "application/atom+xml"},
|
headers={"Content-Type": "application/atom+xml"},
|
||||||
)
|
)
|
||||||
|
|
32
tasks.py
32
tasks.py
|
@ -2,17 +2,49 @@ import asyncio
|
||||||
import io
|
import io
|
||||||
import shutil
|
import shutil
|
||||||
import tarfile
|
import tarfile
|
||||||
|
from collections import namedtuple
|
||||||
from contextlib import contextmanager
|
from contextlib import contextmanager
|
||||||
|
from inspect import getfullargspec
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Generator
|
from typing import Generator
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
|
import invoke # type: ignore
|
||||||
from invoke import Context # type: ignore
|
from invoke import Context # type: ignore
|
||||||
from invoke import run # type: ignore
|
from invoke import run # type: ignore
|
||||||
from invoke import task # type: ignore
|
from invoke import task # type: ignore
|
||||||
|
|
||||||
|
|
||||||
|
def fix_annotations():
|
||||||
|
"""
|
||||||
|
Pyinvoke doesn't accept annotations by default, this fix that
|
||||||
|
Based on: @zelo's fix in https://github.com/pyinvoke/invoke/pull/606
|
||||||
|
Context in: https://github.com/pyinvoke/invoke/issues/357
|
||||||
|
Python 3.11 https://github.com/pyinvoke/invoke/issues/833
|
||||||
|
"""
|
||||||
|
|
||||||
|
ArgSpec = namedtuple("ArgSpec", ["args", "defaults"])
|
||||||
|
|
||||||
|
def patched_inspect_getargspec(func):
|
||||||
|
spec = getfullargspec(func)
|
||||||
|
return ArgSpec(spec.args, spec.defaults)
|
||||||
|
|
||||||
|
org_task_argspec = invoke.tasks.Task.argspec
|
||||||
|
|
||||||
|
def patched_task_argspec(*args, **kwargs):
|
||||||
|
with patch(
|
||||||
|
target="inspect.getargspec", new=patched_inspect_getargspec, create=True
|
||||||
|
):
|
||||||
|
return org_task_argspec(*args, **kwargs)
|
||||||
|
|
||||||
|
invoke.tasks.Task.argspec = patched_task_argspec
|
||||||
|
|
||||||
|
|
||||||
|
fix_annotations()
|
||||||
|
|
||||||
|
|
||||||
@task
|
@task
|
||||||
def generate_db_migration(ctx, message):
|
def generate_db_migration(ctx, message):
|
||||||
# type: (Context, str) -> None
|
# type: (Context, str) -> None
|
||||||
|
|
Loading…
Reference in New Issue