Compare commits

...

2 Commits

Author SHA1 Message Date
Thomas Sileo 673baf0d7f Patch invoke for Python 3.11 support 2022-12-23 09:32:40 +01:00
Thomas Sileo 9c65919070 Tweak feeds 2022-12-23 09:25:50 +01:00
2 changed files with 48 additions and 9 deletions

View File

@ -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, {
"url": LOCAL_ACTOR.url, "name": LOCAL_ACTOR.display_name,
}, "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)
# Atom feeds require a title
if not is_rss:
fe.title(outbox_object.url)
fe.link(href=outbox_object.url) fe.link(href=outbox_object.url)
fe.title(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"},
) )

View File

@ -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