router integration: stuck

apparently, a @staticmethod that Depends on another @staticmethod in the same class is bad
This commit is contained in:
Jörn-Michael Miehe 2023-09-08 19:08:13 +00:00
parent b1748ea0fb
commit af00dafb6c
10 changed files with 96 additions and 99 deletions

View file

@ -48,10 +48,10 @@ class AdventImage:
self, self,
xy: _XY, xy: _XY,
text: str | bytes, text: str | bytes,
font: ImageFont._Font, font: "ImageFont._Font",
anchor: str | None = "mm", anchor: str | None = "mm",
**text_kwargs, **text_kwargs,
) -> Image._Box | None: ) -> "Image._Box | None":
""" """
Koordinaten (links, oben, rechts, unten) des betroffenen Koordinaten (links, oben, rechts, unten) des betroffenen
Rechtecks bestimmen, wenn das Bild mit einem Text Rechtecks bestimmen, wenn das Bild mit einem Text
@ -76,7 +76,7 @@ class AdventImage:
async def get_average_color( async def get_average_color(
self, self,
box: Image._Box, box: "Image._Box",
) -> tuple[int, int, int]: ) -> tuple[int, int, int]:
""" """
Durchschnittsfarbe eines rechteckigen Ausschnitts in Durchschnittsfarbe eines rechteckigen Ausschnitts in
@ -92,7 +92,7 @@ class AdventImage:
self, self,
xy: _XY, xy: _XY,
text: str | bytes, text: str | bytes,
font: ImageFont._Font, font: "ImageFont._Font",
anchor: str | None = "mm", anchor: str | None = "mm",
**text_kwargs, **text_kwargs,
) -> None: ) -> None:

View file

@ -1,7 +1,13 @@
import tomllib
from typing import TypeAlias from typing import TypeAlias
import tomli_w
from fastapi import Depends
from pydantic import BaseModel from pydantic import BaseModel
from .config import Config
from .webdav import WebDAV
class DoorSaved(BaseModel): class DoorSaved(BaseModel):
# Tag, an dem die Tür aufgeht # Tag, an dem die Tür aufgeht
@ -23,3 +29,32 @@ class CalendarConfig(BaseModel):
# Türen für die UI # Türen für die UI
doors: DoorsSaved = [] doors: DoorsSaved = []
@staticmethod
async def get_calendar_config(
cfg: Config = Depends(Config.get_config),
) -> "CalendarConfig":
"""
Kalender Konfiguration lesen
"""
txt = await WebDAV.read_str(path=f"files/{cfg.puzzle.calendar}")
return CalendarConfig.model_validate(tomllib.loads(txt))
async def set_calendar_config(
self,
cfg: Config = Depends(Config.get_config),
) -> None:
"""
Kalender Konfiguration ändern
"""
await WebDAV.write_str(
path=f"files/{cfg.puzzle.calendar}",
content=tomli_w.dumps(
self.model_dump(
exclude_defaults=True,
exclude_unset=True,
)
),
)

View file

@ -1,5 +1,10 @@
import tomllib
from pydantic import BaseModel from pydantic import BaseModel
from .settings import SETTINGS
from .webdav import WebDAV
class User(BaseModel): class User(BaseModel):
name: str name: str
@ -38,3 +43,12 @@ class Config(BaseModel):
admin: User admin: User
server: Server server: Server
puzzle: Puzzle puzzle: Puzzle
@staticmethod
async def get_config() -> "Config":
"""
Globale Konfiguration lesen
"""
txt = await WebDAV.read_str(path=SETTINGS.config_filename)
return Config.model_validate(tomllib.loads(txt))

View file

@ -1,62 +1,20 @@
import tomllib from io import BytesIO
from typing import cast from typing import cast
import tomli_w
from fastapi import Depends from fastapi import Depends
from PIL import Image, ImageFont from PIL import Image, ImageFont
from .advent_image import _XY, AdventImage from .advent_image import _XY, AdventImage
from .calendar_config import CalendarConfig
from .config import Config from .config import Config
from .image_helpers import list_images_auto, load_image from .image_helpers import list_images_auto, load_image
from .sequence_helpers import Random, set_len, shuffle from .sequence_helpers import Random, set_len, shuffle
from .settings import SETTINGS
from .webdav import WebDAV from .webdav import WebDAV
class AllTime: class AllTime:
@staticmethod
async def get_config() -> Config:
"""
Globale Konfiguration lesen
"""
txt = await WebDAV.read_str(path=SETTINGS.config_filename)
return Config.model_validate(tomllib.loads(txt))
@staticmethod
async def get_calendar_config(
cfg: Config = Depends(get_config),
) -> CalendarConfig:
"""
Kalender Konfiguration lesen
"""
txt = await WebDAV.read_str(path=f"files/{cfg.puzzle.calendar}")
return CalendarConfig.model_validate(tomllib.loads(txt))
@staticmethod
async def set_calendar_config(
cal_cfg: CalendarConfig,
cfg: Config = Depends(get_config),
) -> None:
"""
Kalender Konfiguration ändern
"""
await WebDAV.write_str(
path=f"files/{cfg.puzzle.calendar}",
content=tomli_w.dumps(
cal_cfg.model_dump(
exclude_defaults=True,
exclude_unset=True,
)
),
)
@staticmethod @staticmethod
async def shuffle_solution( async def shuffle_solution(
cfg: Config = Depends(get_config), cfg: Config = Depends(Config.get_config),
) -> str: ) -> str:
""" """
Lösung: Reihenfolge zufällig bestimmen Lösung: Reihenfolge zufällig bestimmen
@ -102,7 +60,7 @@ class Today:
async def gen_auto_image( async def gen_auto_image(
day: int, day: int,
images: list[str] = Depends(AllTime.shuffle_images_auto), images: list[str] = Depends(AllTime.shuffle_images_auto),
cfg: Config = Depends(AllTime.get_config), cfg: Config = Depends(Config.get_config),
rnd: Random = Depends(get_random), rnd: Random = Depends(get_random),
part: str = Depends(get_part), part: str = Depends(get_part),
) -> Image.Image: ) -> Image.Image:
@ -115,7 +73,7 @@ class Today:
image = await AdventImage.from_img(img) image = await AdventImage.from_img(img)
font = ImageFont.truetype( font = ImageFont.truetype(
font=await WebDAV.read_bytes(f"files/{cfg.server.font}"), font=BytesIO(await WebDAV.read_bytes(f"files/{cfg.server.font}")),
size=50, size=50,
) )
@ -133,7 +91,7 @@ class Today:
async def get_image( async def get_image(
day: int, day: int,
images: list[str] = Depends(AllTime.shuffle_images_auto), images: list[str] = Depends(AllTime.shuffle_images_auto),
cfg: Config = Depends(AllTime.get_config), cfg: Config = Depends(Config.get_config),
rnd: Random = Depends(get_random), rnd: Random = Depends(get_random),
part: str = Depends(get_part), part: str = Depends(get_part),
) -> Image.Image: ) -> Image.Image:

View file

@ -26,7 +26,7 @@ async def load_image(file_name: str) -> Image.Image:
if not await WebDAV.file_exists(file_name): if not await WebDAV.file_exists(file_name):
raise RuntimeError(f"DAV-File {file_name} does not exist!") raise RuntimeError(f"DAV-File {file_name} does not exist!")
return Image.open(await WebDAV.read_bytes(file_name)) return Image.open(BytesIO(await WebDAV.read_bytes(file_name)))
async def api_return_image(img: Image.Image) -> StreamingResponse: async def api_return_image(img: Image.Image) -> StreamingResponse:

View file

@ -2,14 +2,14 @@ import itertools
import random import random
from typing import Any, Self, Sequence from typing import Any, Self, Sequence
from .depends import AllTime from .config import Config
class Random(random.Random): class Random(random.Random):
@classmethod @classmethod
async def get(cls, bonus_salt: Any = "") -> Self: async def get(cls, bonus_salt: Any = "") -> Self:
cfg = await AllTime.get_config() cfg = await Config.get_config()
return cls(f"{cfg.puzzle.solution}{bonus_salt}{cfg.puzzle.random_pepper}") return cls(f"{cfg.puzzle.solution}{cfg.puzzle.random_pepper}{bonus_salt}")
async def shuffle(seq: Sequence, rnd: random.Random | None = None) -> list: async def shuffle(seq: Sequence, rnd: random.Random | None = None) -> list:

View file

@ -17,8 +17,8 @@ class WebDAV:
} }
) )
@AsyncTTL(time_to_live=SETTINGS.cache_ttl)
@classmethod @classmethod
@AsyncTTL(time_to_live=SETTINGS.cache_ttl)
async def list_files( async def list_files(
cls, cls,
directory: str = "", directory: str = "",
@ -33,8 +33,8 @@ class WebDAV:
return [f"{directory}/{path}" for path in ls if regex.search(path)] return [f"{directory}/{path}" for path in ls if regex.search(path)]
@AsyncTTL(time_to_live=SETTINGS.cache_ttl)
@classmethod @classmethod
@AsyncTTL(time_to_live=SETTINGS.cache_ttl)
async def file_exists(cls, path: str) -> bool: async def file_exists(cls, path: str) -> bool:
""" """
`True`, wenn an Pfad `path` eine Datei existiert `True`, wenn an Pfad `path` eine Datei existiert
@ -42,8 +42,8 @@ class WebDAV:
return cls._webdav_client.check(path) return cls._webdav_client.check(path)
@AsyncTTL(time_to_live=SETTINGS.cache_ttl)
@classmethod @classmethod
@AsyncTTL(time_to_live=SETTINGS.cache_ttl)
async def read_bytes(cls, path: str) -> bytes: async def read_bytes(cls, path: str) -> bytes:
""" """
Datei aus Pfad `path` als bytes laden Datei aus Pfad `path` als bytes laden
@ -51,10 +51,12 @@ class WebDAV:
buffer = BytesIO() buffer = BytesIO()
cls._webdav_client.resource(path).write_to(buffer) cls._webdav_client.resource(path).write_to(buffer)
buffer.seek(0)
return buffer.read() return buffer.read()
@AsyncTTL(time_to_live=SETTINGS.cache_ttl)
@classmethod @classmethod
@AsyncTTL(time_to_live=SETTINGS.cache_ttl)
async def read_str(cls, path: str, encoding="utf-8") -> str: async def read_str(cls, path: str, encoding="utf-8") -> str:
""" """
Datei aus Pfad `path` als string laden Datei aus Pfad `path` als string laden
@ -63,9 +65,9 @@ class WebDAV:
return (await cls.read_bytes(path)).decode(encoding=encoding).strip() return (await cls.read_bytes(path)).decode(encoding=encoding).strip()
@classmethod @classmethod
async def write_buffer(cls, path: str, buffer: BytesIO) -> None: async def write_bytes(cls, path: str, buffer: bytes) -> None:
""" """
Puffer `buffer` in Datei in Pfad `path` schreiben Bytes `buffer` in Datei in Pfad `path` schreiben
""" """
cls._webdav_client.resource(path).read_from(buffer) cls._webdav_client.resource(path).read_from(buffer)
@ -76,7 +78,4 @@ class WebDAV:
String `content` in Datei in Pfad `path` schreiben String `content` in Datei in Pfad `path` schreiben
""" """
buffer = BytesIO(content.encode(encoding=encoding)) await cls.write_bytes(path, content.encode(encoding=encoding))
buffer.seek(0)
await cls.write_buffer(path, buffer)

View file

@ -2,10 +2,11 @@ from datetime import date
from fastapi import APIRouter, Depends, HTTPException, status from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import StreamingResponse from fastapi.responses import StreamingResponse
from PIL import Image
from ..config import Config, get_config from ..core.config import Config
from ._image import AdventImage from ..core.depends import AllTime, Today
from ._misc import api_return_image, get_image, shuffle from ..core.image_helpers import api_return_image
from .user import user_is_admin from .user import user_is_admin
router = APIRouter(prefix="/days", tags=["days"]) router = APIRouter(prefix="/days", tags=["days"])
@ -13,17 +14,16 @@ router = APIRouter(prefix="/days", tags=["days"])
@router.on_event("startup") @router.on_event("startup")
async def startup() -> None: async def startup() -> None:
cfg = await get_config() cfg = await Config.get_config()
print(cfg.puzzle.solution) print(cfg.puzzle.solution)
print("".join(await shuffle(cfg.puzzle.solution)))
shuffled_solution = await AllTime.shuffle_solution(cfg)
print(shuffled_solution)
@router.get("/letter/{index}") @router.get("/part/{day}")
async def get_letter( async def get_part(part: str = Depends(Today.get_part)) -> str:
index: int, return part
cfg: Config = Depends(get_config),
) -> str:
return (await shuffle(cfg.puzzle.solution))[index]
@router.get("/date") @router.get("/date")
@ -45,17 +45,17 @@ async def get_visible_days() -> int:
async def user_can_view( async def user_can_view(
index: int, day: int,
) -> bool: ) -> bool:
return index < await get_visible_days() return day < await get_visible_days()
@router.get( @router.get(
"/image/{index}", "/image/{day}",
response_class=StreamingResponse, response_class=StreamingResponse,
) )
async def get_image_for_day( async def get_image_for_day(
image: AdventImage = Depends(get_image), image: Image.Image = Depends(Today.get_image),
can_view: bool = Depends(user_can_view), can_view: bool = Depends(user_can_view),
is_admin: bool = Depends(user_is_admin), is_admin: bool = Depends(user_is_admin),
) -> StreamingResponse: ) -> StreamingResponse:
@ -66,4 +66,4 @@ async def get_image_for_day(
if not (can_view or is_admin): if not (can_view or is_admin):
raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Wie unhöflich!!!") raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Wie unhöflich!!!")
return await api_return_image(image.img) return await api_return_image(image)

View file

@ -1,15 +1,8 @@
from fastapi import APIRouter, Depends from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse from fastapi.responses import StreamingResponse
from PIL import Image
from ..calendar_config import ( from ..core.calendar_config import CalendarConfig, DoorsSaved
CalendarConfig, from ..core.image_helpers import api_return_image, load_image
DoorsSaved,
get_calendar_config,
set_calendar_config,
)
from ..dav_common import dav_get_file
from ._misc import api_return_image
router = APIRouter(prefix="/general", tags=["general"]) router = APIRouter(prefix="/general", tags=["general"])
@ -19,20 +12,18 @@ router = APIRouter(prefix="/general", tags=["general"])
response_class=StreamingResponse, response_class=StreamingResponse,
) )
async def get_image_for_day( async def get_image_for_day(
cal_cfg: CalendarConfig = Depends(get_calendar_config), cal_cfg: CalendarConfig = Depends(CalendarConfig.get_calendar_config),
) -> StreamingResponse: ) -> StreamingResponse:
""" """
Hintergrundbild laden Hintergrundbild laden
""" """
return await api_return_image( return await api_return_image(await load_image(f"files/{cal_cfg.background}"))
Image.open(await dav_get_file(f"files/{cal_cfg.background}"))
)
@router.get("/doors") @router.get("/doors")
async def get_doors( async def get_doors(
cal_cfg: CalendarConfig = Depends(get_calendar_config), cal_cfg: CalendarConfig = Depends(CalendarConfig.get_calendar_config),
) -> DoorsSaved: ) -> DoorsSaved:
""" """
Türchen lesen Türchen lesen
@ -44,7 +35,7 @@ async def get_doors(
@router.put("/doors") @router.put("/doors")
async def put_doors( async def put_doors(
doors: DoorsSaved, doors: DoorsSaved,
cal_cfg: CalendarConfig = Depends(get_calendar_config), cal_cfg: CalendarConfig = Depends(CalendarConfig.get_calendar_config),
) -> None: ) -> None:
""" """
Türchen setzen Türchen setzen
@ -54,4 +45,4 @@ async def put_doors(
doors, doors,
key=lambda door: door.day, key=lambda door: door.day,
) )
await set_calendar_config(cal_cfg) await cal_cfg.set_calendar_config()

View file

@ -3,7 +3,7 @@ import secrets
from fastapi import APIRouter, Depends, HTTPException, status from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials from fastapi.security import HTTPBasic, HTTPBasicCredentials
from ..config import Config, get_config from ..core.config import Config
router = APIRouter(prefix="/user", tags=["user"]) router = APIRouter(prefix="/user", tags=["user"])
security = HTTPBasic() security = HTTPBasic()
@ -11,7 +11,7 @@ security = HTTPBasic()
async def user_is_admin( async def user_is_admin(
credentials: HTTPBasicCredentials = Depends(security), credentials: HTTPBasicCredentials = Depends(security),
config: Config = Depends(get_config), config: Config = Depends(Config.get_config),
) -> bool: ) -> bool:
username_correct = secrets.compare_digest(credentials.username, config.admin.name) username_correct = secrets.compare_digest(credentials.username, config.admin.name)