[lint] main.py

app/main.py

@@ -1,21 +1,20 @@
 #!/usr/bin/env python3
-import logging
+"""Some application path."""
+from typing import Any
 import sys
 import fastapi
-import httpx
-import json
 
 from fastapi import FastAPI
 from fastapi import Depends
 from fastapi import Request
 from fastapi import Response
 from fastapi.exceptions import HTTPException
-from sqlalchemy.util import monkeypatch_proxied_specials
 from starlette.responses import JSONResponse
-from typing import Any
 
 from loguru import logger
+from sqlalchemy import select
 
+from app import models
 from app.utils import precheck
 from app.database import get_db_session
 from app.config import DEBUG
@@ -26,11 +25,11 @@ from app.config import ID
 from app.config import USERNAME
 from app.config import VERSION
 from app.database import AsyncSession
-from app.database import get_db_session
 from app.boxes import save_incoming
+from app.activitypub import VisibilityEnum
+from app.boxes import _send_create
+from app.orgpython import to_html
 
-from sqlalchemy import select
-from app import models
 
 
 def _check_0rtt_early_data(request: Request) -> None:
@@ -46,18 +45,20 @@ logger.remove()
 logger.add(sys.stdout, level="DEBUG" if DEBUG else "INFO")
 logger.add("output.log", level="DEBUG")
 
-class ActivityPubResponse(JSONResponse):
+class ActivityPubResponse(JSONResponse): #pylint: disable=too-few-public-methods
+    """Simple wrap JSONresponse return ActivityPub response."""
     media_type = "application/activity+json"
 
 
 def is_ap_requested(req: Request) -> bool:
+    """Check resquest is ActivityPub request."""
     accept_value = req.headers.get("accept")
     if not accept_value:
         return False
-    for i in {
+    for i in [
         "application/activity+json",
         "application/ld+json",
-    }:
+    ]:
         if accept_value.startswith(i):
             return True
 
@@ -68,72 +69,72 @@ def is_ap_requested(req: Request) -> bool:
 async def index(
         request: Request
 ):
+    """Return index page."""
     if is_ap_requested(request):
         return ActivityPubResponse(ME)
 
     return ME
 
+
 @app.post("/inbox")
 async def inbox(
     request: Request,
     db_session: AsyncSession = Depends(get_db_session),
     httpsig_checker = Depends(precheck.inbox_prechecker),
 ) -> Response:
+    """ActivityPub inbox endpoint."""
     payload = await request.json()
 
-    if httpsig_checker:
-        if await save_incoming(db_session, payload):
-            return Response(status_code=202)
-        else:
-            return Response(
-                status_code=406,
-                content="invalid activitypub object"
-            )
-    else:
+    if not httpsig_checker:
         return Response(
             status_code=406,
             content="invalid http-sig"
         )
 
+    if not await save_incoming(db_session, payload):
+        return Response(
+            status_code=406,
+            content="invalid activitypub object"
+        )
+
+    return Response(status_code=202)
+
 
 @app.post("/outbox")
 async def outbox(
         request: Request,
         db_session: AsyncSession = Depends(get_db_session),
 ) -> Response:
+    """ActivityPub outbox endpoint, now only process client post request."""
     payload = await request.body()
     content = payload.decode("utf-8")
     logger.info(content)
     post_token = request.headers.get("Authorization")
 
-    if POST_TOKEN is not None \
-    and POST_TOKEN == post_token.split(" ")[1]: # type: ignore
-        logger.info("True token")
-
-        from app.activitypub import VisibilityEnum
-        from app.boxes import _send_create
-        from app.orgpython import to_html
-
-        content = to_html(content).replace("\n", "")
-
-        await _send_create(
-            db_session,
-            "Note",
-            content,
-            VisibilityEnum.PUBLIC
-        )
-        return Response(status_code=200)
-    else:
+    if POST_TOKEN is None \
+    and POST_TOKEN != post_token.split(" ")[1]: # type: ignore
         logger.warning("Non-valid post token!")
         return Response(status_code=406)
 
+    logger.info("True token")
+
+    content = to_html(content).replace("\n", "")
+
+    await _send_create(
+        db_session,
+        "Note",
+        content,
+        VisibilityEnum.PUBLIC
+    )
+    return Response(status_code=200)
+
 
 @app.get("/tail/{public_id}")
 async def outbox_activity_by_public_id(
     public_id: str,
-    request: Request,
     db_session: AsyncSession = Depends(get_db_session),
 ) -> ActivityPubResponse:
+    """Return note page."""
     outbox_object = (
         await db_session.execute(
             select(models.OutboxObject).where(
@@ -178,8 +179,10 @@ async def wellknown_webfinger(resource: str) -> JSONResponse:
         headers={"Access-Control-Allow-Origin": "*"},
     )
 
+
 @app.get("/.well-known/nodeinfo")
 async def well_known_nodeinfo() -> dict[str, Any]:
+    """Exposes nodeinfo path."""
     return {
         "links": [
             {
@@ -191,9 +194,8 @@ async def well_known_nodeinfo() -> dict[str, Any]:
 
 
 @app.get("/nodeinfo/2.0")
-async def nodeinfo(
-    db_session: AsyncSession = Depends(get_db_session),
-):
+async def nodeinfo():
+    """Return site nodeinfo."""
     return JSONResponse(
         {
             "version": "2.0",