Html templates

.github/workflows/python-check.yml

@@ -67,6 +67,7 @@ jobs:
           TESTING: 1
           MOVIE_CATALOG__REDIS__CONNECTION__HOST: redis
           MOVIE_CATALOG__REDIS__CONNECTION__PORT: 6379
+          MOVIE_CATALOG__SESSION__SECRET_KEY: "abc"
 
       - name: Upload artefacts
         uses: actions/upload-artifact@v4

movie_catalog/api/api_v1/movie_catalog/dependencies.py

@@ -4,47 +4,21 @@
 from fastapi import Depends, HTTPException, status
 from fastapi.security import (
     HTTPAuthorizationCredentials,
-    HTTPBasic,
     HTTPBasicCredentials,
     HTTPBearer,
 )
 
-from movie_catalog.schemas.movie_catalog import Movie
-
-from ..auth.services import redis_tokens, redis_users
-from .crud import storage
+from movie_catalog.dependencies.auth import user_basic_auth, validate_basic_auth
+from movie_catalog.services.auth import redis_tokens
 
 logger = logging.getLogger(__name__)
 
-UNSAFE_METHOD = frozenset(
-    {
-        "POST",
-        "PUT",
-        "PATCH",
-        "DELETE",
-    }
-)
 static_api_token = HTTPBearer(
     scheme_name="Static API token",
     description="Your **Static API token** from the developer portal. [Read more](#)",
     auto_error=False,
 )
 
-user_basic_auth = HTTPBasic(
-    scheme_name="Basic auth",
-    description="Basic username + password auth",
-    auto_error=False,
-)
-
-
-def prefetch_film(slug: str) -> Movie:
-    films: Movie | None = storage.get_by_slug(slug)
-    if films:
-        return films
-    raise HTTPException(
-        status_code=status.HTTP_404_NOT_FOUND, detail=f"Movie {slug!r} not found"
-    )
-
 
 def validate_api_token(api_token: HTTPAuthorizationCredentials) -> None:
     logger.debug("API token: %s", api_token)
@@ -71,29 +45,6 @@ def api_token_required(
     validate_api_token(api_token=api_token)
 
 
-def validate_basic_auth(credentials: HTTPBasicCredentials | None) -> None:
-    logger.debug("User credentials: %s", credentials)
-
-    if credentials and redis_users.validate_user_password(
-        username=credentials.username, password=credentials.password
-    ):
-        return
-
-    raise HTTPException(
-        status_code=status.HTTP_401_UNAUTHORIZED,
-        detail="Invalid username or password.",
-        headers={"WWW-Authenticate": "Basic"},
-    )
-
-
-def user_basic_auth_required(
-    credentials: Annotated[
-        HTTPBasicCredentials | None, Depends(user_basic_auth)
-    ] = None,
-) -> None:
-    validate_basic_auth(credentials=credentials)
-
-
 def api_token_or_user_basic_auth_required(
     api_token: Annotated[
         HTTPAuthorizationCredentials | None, Depends(static_api_token)

movie_catalog/api/api_v1/movie_catalog/views/details_view.py

@@ -1,18 +1,16 @@
-from typing import Annotated
-
 from fastapi import APIRouter, Depends, status
 
-from movie_catalog.api.api_v1.movie_catalog.crud import storage
 from movie_catalog.api.api_v1.movie_catalog.dependencies import (
     api_token_or_user_basic_auth_required,
-    prefetch_film,
 )
+from movie_catalog.dependencies.movie_catalog import MovieBySlug
 from movie_catalog.schemas.movie_catalog import (
     Movie,
     MoviePartialUpdate,
     MovieRead,
     MovieUpdate,
 )
+from movie_catalog.storage.movie_catalog.crud import storage
 
 router = APIRouter(
     prefix="/{slug}",
@@ -50,8 +48,6 @@
     },
 )
 
-MovieBySlug = Annotated[Movie, Depends(prefetch_film)]
-
 
 @router.get("/", response_model=MovieRead)
 def get_movie(movie: MovieBySlug) -> Movie:

movie_catalog/api/api_v1/movie_catalog/views/list_view.py

@@ -2,14 +2,14 @@
 
 from fastapi import APIRouter, Depends, HTTPException, status
 
-from movie_catalog.api.api_v1.movie_catalog.crud import (
-    MovieCatalogAlreadyExists,
-    storage,
-)
 from movie_catalog.api.api_v1.movie_catalog.dependencies import (
     api_token_or_user_basic_auth_required,
 )
+from movie_catalog.exceptions import MovieAlreadyExists
 from movie_catalog.schemas.movie_catalog import Movie, MovieCreate, MovieRead
+from movie_catalog.storage.movie_catalog.crud import (
+    storage,
+)
 
 router: APIRouter = APIRouter(
     prefix="/movies",
@@ -70,7 +70,7 @@ def add_movie(
 ) -> Movie:
     try:
         return storage.create_or_rise_if_exists(movie_create)
-    except MovieCatalogAlreadyExists:
+    except MovieAlreadyExists:
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
             detail=f"Movie with slug <{movie_create.slug}> already exists.",

movie_catalog/app_lifespan.py

@@ -3,10 +3,15 @@
 
 from fastapi import FastAPI
 
+from movie_catalog.config import settings
+from movie_catalog.storage.movie_catalog import MovieCatalogStorage
+
 
 @asynccontextmanager
 async def lifespan(
     app: FastAPI,  # noqa: ARG001
 ) -> AsyncIterator[None]:
-
+    app.state.movie_catalog_storage = MovieCatalogStorage(
+        hast_name=settings.redis.collections.movie_catalog_hash,
+    )
     yield

movie_catalog/commands/tokens.py

@@ -2,9 +2,9 @@
 from typing import Annotated
 
 import typer
-from api.api_v1.auth.services import redis_tokens
 from rich import print
 from rich.markdown import Markdown
+from services.auth import redis_tokens
 
 app = typer.Typer(
     name="token",

movie_catalog/config.default.yaml

@@ -1,3 +1,5 @@
 movie-catalog:
   logging:
     log_level_name: INFO
+  session:
+    secret_key:

movie_catalog/config.py

@@ -55,6 +55,10 @@ class RedisConfig(BaseModel):
     collections: RedisCollectionsNamesConfig = RedisCollectionsNamesConfig()
 
 
+class SessionConfig(BaseModel):
+    secret_key: str
+
+
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
         case_sensitive=False,
@@ -103,6 +107,7 @@ def settings_customise_sources(
 
     logging: LoggingConfig = LoggingConfig()
     redis: RedisConfig = RedisConfig()
+    session: SessionConfig
 
 
 settings = Settings()

movie_catalog/dependencies/auth.py

@@ -0,0 +1,52 @@
+import logging
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBasic, HTTPBasicCredentials
+from starlette.requests import Request
+
+from movie_catalog.services.auth import redis_users
+
+logger = logging.getLogger(__name__)
+
+
+UNSAFE_METHOD = frozenset(
+    {
+        "POST",
+        "PUT",
+        "PATCH",
+        "DELETE",
+    }
+)
+user_basic_auth = HTTPBasic(
+    scheme_name="Basic auth",
+    description="Basic username + password auth",
+    auto_error=False,
+)
+
+
+def validate_basic_auth(credentials: HTTPBasicCredentials | None) -> None:
+    logger.debug("User credentials: %s", credentials)
+
+    if credentials and redis_users.validate_user_password(
+        username=credentials.username, password=credentials.password
+    ):
+        return
+
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid username or password.",
+        headers={"WWW-Authenticate": "Basic"},
+    )
+
+
+def user_basic_auth_required_for_unsafe_methods(
+    request: Request,
+    credentials: Annotated[
+        HTTPBasicCredentials | None, Depends(user_basic_auth)
+    ] = None,
+) -> None:
+    if request.method not in UNSAFE_METHOD:
+        return
+
+    validate_basic_auth(credentials=credentials)

movie_catalog/dependencies/movie_catalog.py

@@ -0,0 +1,32 @@
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, Request
+from starlette import status
+
+from movie_catalog.schemas.movie_catalog import Movie
+from movie_catalog.storage.movie_catalog import MovieCatalogStorage
+from movie_catalog.storage.movie_catalog.crud import storage
+
+
+def get_movie_catalog_storage(
+    request: Request,
+) -> MovieCatalogStorage:
+    return request.app.state.movie_catalog_storage  # type: ignore[no-any-return]
+
+
+GetMovieCatalogStorage = Annotated[
+    MovieCatalogStorage,
+    Depends(get_movie_catalog_storage),
+]
+
+
+def prefetch_movie(slug: str) -> Movie:
+    movie: Movie | None = storage.get_by_slug(slug)
+    if movie:
+        return movie
+    raise HTTPException(
+        status_code=status.HTTP_404_NOT_FOUND, detail=f"Movie {slug!r} not found"
+    )
+
+
+MovieBySlug = Annotated[Movie, Depends(prefetch_movie)]

movie_catalog/exceptions.py

@@ -0,0 +1,10 @@
+class MovieCatalogBaseError(Exception):
+    """
+    Base exception class for Movie Catalog
+    """
+
+
+class MovieAlreadyExists(MovieCatalogBaseError):
+    """
+    Exception raised when a movie already exists
+    """

movie_catalog/main.py

@@ -1,11 +1,12 @@
 import logging
 
 from fastapi import FastAPI
+from starlette.middleware.sessions import SessionMiddleware
 
 from movie_catalog.api import router as api_router
-from movie_catalog.api.main_view import router as main_router
 from movie_catalog.app_lifespan import lifespan
 from movie_catalog.config import settings
+from movie_catalog.rest import router as rest_router
 
 logging.basicConfig(
     format=settings.logging.log_format,
@@ -19,5 +20,9 @@
     version="1.0",
     lifespan=lifespan,
 )
-app.include_router(main_router)
+app.add_middleware(
+    SessionMiddleware,
+    secret_key=settings.session.secret_key,
+)
+app.include_router(rest_router)
 app.include_router(api_router)

movie_catalog/manage.bat

@@ -1,2 +1,2 @@
 @echo off
-uv run --script manage.py %*
+uv run --script movie_catalog/manage.py %*

movie_catalog/misc/flash_messages.py

@@ -0,0 +1,26 @@
+from typing import TypedDict
+
+from fastapi import Request
+
+FLASHED_MESSAGES_KEY = "_flashed_messages"
+
+
+class Message(TypedDict):
+    message: str
+    category: str
+
+
+def flash(request: Request, message: str, category: str = "info") -> None:
+    if FLASHED_MESSAGES_KEY not in request.session:
+        request.session[FLASHED_MESSAGES_KEY] = []
+
+    request.session[FLASHED_MESSAGES_KEY].append(
+        Message(
+            message=message,
+            category=category,
+        )
+    )
+
+
+def get_flashed_messages(request: Request) -> list[Message]:
+    return request.session.pop(FLASHED_MESSAGES_KEY, [])  # type: ignore[no-any-return]

movie_catalog/rest/__init__.py

@@ -0,0 +1,10 @@
+from fastapi import APIRouter
+
+from .main_view import router as main_router
+from .movie_catalog import router as movie_catalog_router
+
+router = APIRouter(
+    include_in_schema=False,
+)
+router.include_router(main_router)
+router.include_router(movie_catalog_router)