kopia lustrzana https://git.sr.ht/~tsileo/microblog.pub
Porównaj commity
2 Commity
c506299089
...
673baf0d7f
Autor | SHA1 | Data |
---|---|---|
Thomas Sileo | 673baf0d7f | |
Thomas Sileo | 9c65919070 |
21
app/main.py
21
app/main.py
|
@ -1640,23 +1640,26 @@ async def json_feed(
|
|||
}
|
||||
)
|
||||
result = {
|
||||
"version": "https://jsonfeed.org/version/1",
|
||||
"version": "https://jsonfeed.org/version/1.1",
|
||||
"title": f"{LOCAL_ACTOR.display_name}'s microblog'",
|
||||
"home_page_url": LOCAL_ACTOR.url,
|
||||
"feed_url": BASE_URL + "/feed.json",
|
||||
"author": {
|
||||
"authors": [
|
||||
{
|
||||
"name": LOCAL_ACTOR.display_name,
|
||||
"url": LOCAL_ACTOR.url,
|
||||
},
|
||||
}
|
||||
],
|
||||
"items": data,
|
||||
}
|
||||
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
|
||||
|
||||
|
||||
async def _gen_rss_feed(
|
||||
db_session: AsyncSession,
|
||||
is_rss: bool,
|
||||
):
|
||||
fg = FeedGenerator()
|
||||
fg.id(BASE_URL + "/feed.rss")
|
||||
|
@ -1687,8 +1690,12 @@ async def _gen_rss_feed(
|
|||
|
||||
fe = fg.add_entry()
|
||||
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.link(href=outbox_object.url)
|
||||
fe.description(content)
|
||||
fe.content(content)
|
||||
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),
|
||||
) -> 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"},
|
||||
)
|
||||
|
||||
|
@ -1711,6 +1718,6 @@ async def atom_feed(
|
|||
db_session: AsyncSession = Depends(get_db_session),
|
||||
) -> 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"},
|
||||
)
|
||||
|
|
32
tasks.py
32
tasks.py
|
@ -2,17 +2,49 @@ import asyncio
|
|||
import io
|
||||
import shutil
|
||||
import tarfile
|
||||
from collections import namedtuple
|
||||
from contextlib import contextmanager
|
||||
from inspect import getfullargspec
|
||||
from pathlib import Path
|
||||
from typing import Generator
|
||||
from typing import Optional
|
||||
from unittest.mock import patch
|
||||
|
||||
import httpx
|
||||
import invoke # type: ignore
|
||||
from invoke import Context # type: ignore
|
||||
from invoke import run # 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
|
||||
def generate_db_migration(ctx, message):
|
||||
# type: (Context, str) -> None
|
||||
|
|
Ładowanie…
Reference in New Issue