jonas/kirche-einfach
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

chore: enforce ruff lint and format standards across backend

Browse source
This commit is contained in:
Jonas Miederer 2026-09-14 08:10:14 +02:00
parent a8e312a4f8
commit a6cc585dd0
39 changed files with 459 additions and 559 deletions
Download patch file Download diff file Expand all files Collapse all files

33
backend/app/auth/api.py
View file

@ -1,5 +1,6 @@
"""Access + auth HTTP API (magic-link login, session cookie)."""
from typing import Annotated
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
@ -68,11 +69,9 @@ def _set_session_cookie(response: Response, session_id: str) -> None:
@access_router.post("/requests", response_model=AccessRequestRead, status_code=201)
def create_request(data: AccessRequestCreate, session: Session = Depends(get_db)):
def create_request(data: AccessRequestCreate, session: Annotated[Session, Depends(get_db)]):
try:
request = service.create_access_request(
session, data.name, data.email, data.reason
)
request = service.create_access_request(session, data.name, data.email, data.reason)
except service.RateLimitedError as exc:
raise HTTPException(status_code=429, detail=str(exc)) from exc
mail.send_access_notification(str(request.id), request.name, request.email)
@ -81,9 +80,9 @@ def create_request(data: AccessRequestCreate, session: Session = Depends(get_db)
@access_router.get("/requests", response_model=list[AccessRequestRead])
def list_requests(
session: Annotated[Session, Depends(get_db)],
admin: Annotated[User, Depends(require_admin)],
status: str | None = Query(default=None),
session: Session = Depends(get_db),
admin: User = Depends(require_admin),
):
return [_read_request(r) for r in service.list_access_requests(session, status)]
@ -91,13 +90,11 @@ def list_requests(
@access_router.post("/requests/{request_id}/approve", response_model=AccessDecisionRead)
def approve_request(
request_id: UUID,
session: Session = Depends(get_db),
admin: User = Depends(require_admin),
session: Annotated[Session, Depends(get_db)],
admin: Annotated[User, Depends(require_admin)],
):
try:
request, _user, token = service.approve_access_request(
session, request_id, admin.email
)
request, _user, token = service.approve_access_request(session, request_id, admin.email)
except service.NotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
except service.ConflictError as exc:
@ -110,8 +107,8 @@ def approve_request(
@access_router.post("/requests/{request_id}/reject", response_model=AccessRequestRead)
def reject_request(
request_id: UUID,
session: Session = Depends(get_db),
admin: User = Depends(require_admin),
session: Annotated[Session, Depends(get_db)],
admin: Annotated[User, Depends(require_admin)],
):
try:
request = service.reject_access_request(session, request_id, admin.email)
@ -125,7 +122,7 @@ def reject_request(
@auth_router.post("/magic/request", response_model=MagicRequestRead)
def request_magic(
data: MagicRequestRequest,
session: Session = Depends(get_db),
session: Annotated[Session, Depends(get_db)],
):
"""Email-first login: allowed emails get a magic link mailed.
@ -146,7 +143,7 @@ def request_magic(
def consume_magic(
data: MagicConsumeRequest,
response: Response,
session: Session = Depends(get_db),
session: Annotated[Session, Depends(get_db)],
):
try:
user_session = service.consume_magic_token(session, data.token)
@ -164,7 +161,7 @@ def consume_magic(
@auth_router.get("/me", response_model=MeRead)
def read_me(user: User = Depends(get_current_user)):
def read_me(user: Annotated[User, Depends(get_current_user)]):
return _read_me(user)
@ -172,7 +169,7 @@ def read_me(user: User = Depends(get_current_user)):
def logout(
request: Request,
response: Response,
session: Session = Depends(get_db),
session: Annotated[Session, Depends(get_db)],
):
# Revoke the server-side row when the cookie maps to a live session;
# always clear the client cookie.
@ -184,7 +181,7 @@ def logout(
@auth_router.post("/admin-bootstrap", response_model=MessageRead, status_code=201)
def admin_bootstrap(data: AdminBootstrapRequest, session: Session = Depends(get_db)):
def admin_bootstrap(data: AdminBootstrapRequest, session: Annotated[Session, Depends(get_db)]):
try:
_user, token = service.bootstrap_admin(session, data.email)
except service.ConflictError as exc:

13
backend/app/auth/deps.py
View file

@ -3,6 +3,7 @@
from __future__ import annotations
from collections.abc import Iterator
from typing import Annotated
from fastapi import Depends, HTTPException, Request
from sqlmodel import Session
@ -21,7 +22,7 @@ def get_db() -> Iterator[Session]:
def get_optional_user(
request: Request, session: Session = Depends(get_db)
request: Request, session: Annotated[Session, Depends(get_db)]
) -> User | None:
session_id = request.cookies.get(SESSION_COOKIE)
if not session_id:
@ -32,24 +33,20 @@ def get_optional_user(
return service.get_user(session, record.user_id)
def get_current_user(
request: Request, session: Session = Depends(get_db)
) -> User:
def get_current_user(request: Request, session: Annotated[Session, Depends(get_db)]) -> User:
user = get_optional_user(request, session)
if user is None:
raise HTTPException(status_code=401, detail="not authenticated")
return user
def require_admin(user: User = Depends(get_current_user)) -> User:
def require_admin(user: Annotated[User, Depends(get_current_user)]) -> User:
if user.role != "admin":
raise HTTPException(status_code=403, detail="admin only")
return user
def resolve_tenant(
user: User | None, explicit: str | None, default: str = "default"
) -> str:
def resolve_tenant(user: User | None, explicit: str | None, default: str = "default") -> str:
"""Tenant comes from the session; the body/query param is a deprecated
fallback for unauthenticated callers."""
if user is not None:

19
backend/app/auth/mail.py
View file

@ -42,8 +42,8 @@ def _page(title: str, intro: str, button_label: str, button_url: str | None, not
f'<td bgcolor="{BLUE}" style="border-radius:8px;">'
f'<a href="{button_url}" '
'style="display:inline-block;padding:12px 28px;font-family:-apple-system,'
'Segoe UI,Roboto,Helvetica,Arial,sans-serif;font-size:15px;'
'font-weight:600;color:#ffffff;text-decoration:none;'
"Segoe UI,Roboto,Helvetica,Arial,sans-serif;font-size:15px;"
"font-weight:600;color:#ffffff;text-decoration:none;"
f'border-radius:8px;">{button_label}</a></td></tr></table>'
)
return f"""<!DOCTYPE html>
@ -106,14 +106,10 @@ def _branded_mail(
message["From"] = settings.mail_from or settings.contact_to or to_email
if settings.contact_to:
message["Reply-To"] = settings.contact_to
message.set_content(
f"{headline}\n\n{intro}\n\n{button_label}:\n{button_url}\n\n{note}\n"
)
message.set_content(f"{headline}\n\n{intro}\n\n{button_label}:\n{button_url}\n\n{note}\n")
# Email-safe HTML: tables + inline CSS; some clients strip <style>, so
# the palette is inlined via the f-string constants above.
message.add_alternative(
_page(headline, intro, button_label, button_url, note), subtype="html"
)
message.add_alternative(_page(headline, intro, button_label, button_url, note), subtype="html")
return message
@ -149,7 +145,7 @@ def send_magic_link(to_email: str, link: str) -> bool:
)
try:
_send(message)
except Exception: # noqa: BLE001 - delivery is best effort, see docstring
except (OSError, smtplib.SMTPException):
logger.warning("magic-link delivery failed for %s", to_email)
return False
logger.info("magic link mailed to %s", to_email)
@ -166,8 +162,7 @@ def send_access_notification(request_id: str, name: str, email: str) -> bool:
to_email=settings.contact_to,
headline="Neue Zugangsanfrage",
intro=(
f"{name} ({email}) hat Zugang angefordert und wartet auf "
"Freigabe in der Verwaltung."
f"{name} ({email}) hat Zugang angefordert und wartet auf Freigabe in der Verwaltung."
),
button_label="In der Verwaltung prüfen",
button_url=f"{settings.frontend_url.rstrip('/')}/app/admin",
@ -175,7 +170,7 @@ def send_access_notification(request_id: str, name: str, email: str) -> bool:
)
try:
_send(message)
except Exception: # noqa: BLE001 - best effort, request is already stored
except (OSError, smtplib.SMTPException):
logger.warning("access-request notification failed for %s", request_id)
return False
return True

18
backend/app/auth/service.py
View file

@ -9,7 +9,7 @@ from __future__ import annotations
import hashlib
import logging
import secrets
from datetime import datetime, timedelta, timezone
from datetime import UTC, datetime, timedelta
from uuid import UUID
from sqlmodel import Session, func, select
@ -61,7 +61,7 @@ class RateLimitedError(AuthError):
def utcnow() -> datetime:
return datetime.now(timezone.utc)
return datetime.now(UTC)
def normalize_email(email: str) -> str:
@ -85,9 +85,7 @@ def get_user(session: Session, user_id: UUID) -> User | None:
def get_user_by_email(session: Session, email: str) -> User | None:
return session.exec(
select(User).where(User.email == normalize_email(email))
).first()
return session.exec(select(User).where(User.email == normalize_email(email))).first()
def user_count(session: Session) -> int:
@ -223,9 +221,7 @@ def approve_access_request(
return request, user, token
def reject_access_request(
session: Session, request_id: UUID, decided_by: str
) -> AccessRequest:
def reject_access_request(session: Session, request_id: UUID, decided_by: str) -> AccessRequest:
request = get_access_request(session, request_id)
if request.status != "pending":
raise ConflictError(f"access request {request_id} already {request.status}")
@ -242,16 +238,14 @@ def reject_access_request(
def _aware(value: datetime) -> datetime:
# SQLite returns naive datetimes; stored values are UTC.
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value.replace(tzinfo=UTC)
return value
def consume_magic_token(session: Session, token: str) -> UserSession:
"""Single-use consume → session row. used_at is set before creation."""
record = session.exec(
select(MagicLinkToken).where(
MagicLinkToken.token_hash == hash_token(token.strip())
)
select(MagicLinkToken).where(MagicLinkToken.token_hash == hash_token(token.strip()))
).first()
if record is None:
raise InvalidTokenError("unknown token")