Apply better-py skill rules

.docker/configure_keycloak_client.py

@@ -4,7 +4,6 @@
 import logging
 import os
 import time
-from typing import Any
 
 import httpx
 
@@ -42,7 +41,7 @@ def _get_keycloak_client_representation(
     client: httpx.Client,
     realm: str,
     client_id: str,
-) -> dict[str, Any]:
+) -> dict[str, object]:
     response = client.get(
         f"/admin/realms/{realm}/clients",
         params={"clientId": client_id},
@@ -68,9 +67,7 @@ def configure_keycloak_client() -> None:
             realm,
             client_id,
         )
-        representation["redirectUris"] = [
-            f"{app_public_url}/docs/oauth2-redirect"
-        ]
+        representation["redirectUris"] = [f"{app_public_url}/docs/oauth2-redirect"]
         representation["webOrigins"] = [app_public_url]
         response = client.put(
             f"/admin/realms/{realm}/clients/{representation['id']}",

Makefile

@@ -35,10 +35,10 @@ check: lint test-unit
 lint: lint-ruff lint-format lint-ty lint-imports lint-typos
 
 lint-ruff:
-	uv run --group lint --all-packages ruff check app tests alembic seed scenario-tests
+	uv run --group lint --all-packages ruff check app tests alembic seed scenario-tests .docker
 
 lint-format:
-	uv run --group lint --all-packages ruff format --check app tests alembic seed scenario-tests
+	uv run --group lint --all-packages ruff format --check app tests alembic seed scenario-tests .docker
 
 lint-ty:
 	uv run --group lint --group migration --group test --all-packages ty check --error-on-warning app tests alembic seed scenario-tests
@@ -50,8 +50,8 @@ lint-typos:
 	uvx typos
 
 format:
-	uv run --group lint --all-packages ruff check --fix app tests alembic seed scenario-tests
-	uv run --group lint --all-packages ruff format app tests alembic seed scenario-tests
+	uv run --group lint --all-packages ruff check --fix app tests alembic seed scenario-tests .docker
+	uv run --group lint --all-packages ruff format app tests alembic seed scenario-tests .docker
 
 run:
 	@if [ ! -f .env ]; then \
@@ -90,4 +90,4 @@ test-cov:
 	uv run --group test pytest --cov --cov-report=term-missing
 
 openapi-snapshot:
-	UPDATE_OPENAPI_SNAPSHOT=1 uv run --group test pytest tests/unit/test_openapi_contract.py -q
+	uv run --group test pytest tests/unit/test_openapi_contract.py -q --update-openapi-snapshot

app/auth/__init__.py

@@ -5,7 +5,7 @@
     authenticate_request,
     get_token_validator,
 )
-from app.auth.openapi import OIDCOpenAPIFastAPI
+from app.auth.openapi import OIDCOpenAPIFastAPI, OpenAPIDocument
 from app.auth.token_validator import AccessTokenValidator
 
 __all__: list[str] = [
@@ -14,6 +14,7 @@
     "AccessTokenValidator",
     "AuthClaims",
     "OIDCOpenAPIFastAPI",
+    "OpenAPIDocument",
     "authenticate_request",
     "get_token_validator",
     "require_role",

app/auth/openapi.py

@@ -10,6 +10,8 @@
 from app.exceptions import PROBLEM_JSON_MEDIA_TYPE, ProblemDetails
 from app.settings import OIDCMetadata, Settings
 
+type OpenAPIDocument = dict[str, Any]
+
 _JSON_MEDIA_TYPE = "application/json"
 _ERROR_STATUS_PREFIXES = ("4", "5")
 _GENERATED_VALIDATION_STATUS = str(HTTPStatus.UNPROCESSABLE_CONTENT.value)
@@ -19,7 +21,7 @@
 _VALIDATION_DESCRIPTION = "Validation Error"
 
 
-def _use_problem_json_for_error_responses(schema: dict[str, Any]) -> None:
+def _use_problem_json_for_error_responses(schema: OpenAPIDocument) -> None:
     for operations in schema.get("paths", {}).values():
         for operation in operations.values():
             for status, response in operation.get("responses", {}).items():
@@ -30,7 +32,7 @@ def _use_problem_json_for_error_responses(schema: dict[str, Any]) -> None:
                     content[PROBLEM_JSON_MEDIA_TYPE] = content.pop(_JSON_MEDIA_TYPE)
 
 
-def _is_generated_validation_response(response: dict[str, Any]) -> bool:
+def _is_generated_validation_response(response: OpenAPIDocument) -> bool:
     return response == {
         "description": _VALIDATION_DESCRIPTION,
         "content": {
@@ -39,7 +41,7 @@ def _is_generated_validation_response(response: dict[str, Any]) -> bool:
     }
 
 
-def _add_problem_details(response: dict[str, Any]) -> None:
+def _add_problem_details(response: OpenAPIDocument) -> None:
     problem_ref = {"$ref": _PROBLEM_DETAILS_REF}
     media = response.setdefault("content", {}).setdefault(PROBLEM_JSON_MEDIA_TYPE, {})
     declared = media.get("schema")
@@ -51,7 +53,7 @@ def _add_problem_details(response: dict[str, Any]) -> None:
 
 
 def _drop_unreferenced_components(
-    schema: dict[str, Any], names: tuple[str, ...]
+    schema: OpenAPIDocument, names: tuple[str, ...]
 ) -> None:
     components = schema.get("components", {}).get("schemas", {})
     for name in names:
@@ -61,30 +63,40 @@ def _drop_unreferenced_components(
             del components[name]
 
 
-def _declare_validation_responses(schema: dict[str, Any], status: HTTPStatus) -> None:
+def _merged_validation_description(declared: OpenAPIDocument) -> str:
+    description = declared.get("description")
+    if not description:
+        return _VALIDATION_DESCRIPTION
+    return f"{description} or {_VALIDATION_DESCRIPTION}"
+
+
+def _declare_operation_validation_response(
+    operation: OpenAPIDocument, status: HTTPStatus
+) -> bool:
+    responses = operation.setdefault("responses", {})
+    generated = responses.get(_GENERATED_VALIDATION_STATUS)
+    if generated is not None and _is_generated_validation_response(generated):
+        del responses[_GENERATED_VALIDATION_STATUS]
+    elif not operation.get("parameters") and "requestBody" not in operation:
+        return False
+
+    declared = responses.get(str(status.value))
+    if declared is None:
+        declared = responses[str(status.value)] = {"description": status.phrase}
+    elif "$ref" in declared:
+        return True
+    else:
+        declared["description"] = _merged_validation_description(declared)
+    _add_problem_details(declared)
+    return True
+
+
+def _declare_validation_responses(schema: OpenAPIDocument, status: HTTPStatus) -> None:
     rewritten = False
     for operations in schema.get("paths", {}).values():
         for operation in operations.values():
-            responses = operation.setdefault("responses", {})
-            generated = responses.get(_GENERATED_VALIDATION_STATUS)
-            if generated is not None and _is_generated_validation_response(generated):
-                del responses[_GENERATED_VALIDATION_STATUS]
-            elif not operation.get("parameters") and "requestBody" not in operation:
-                continue
-            rewritten = True
-            declared = responses.get(str(status.value))
-            if declared is None:
-                declared = responses[str(status.value)] = {"description": status.phrase}
-            elif "$ref" in declared:
-                continue
-            else:
-                description = declared.get("description")
-                declared["description"] = (
-                    f"{description} or {_VALIDATION_DESCRIPTION}"
-                    if description
-                    else _VALIDATION_DESCRIPTION
-                )
-            _add_problem_details(declared)
+            if _declare_operation_validation_response(operation, status):
+                rewritten = True
     if not rewritten:
         return
     _drop_unreferenced_components(schema, _GENERATED_VALIDATION_SCHEMAS)
@@ -94,7 +106,7 @@ def _declare_validation_responses(schema: dict[str, Any], status: HTTPStatus) ->
 
 
 def _use_discovered_oidc_endpoints(
-    schema: dict[str, Any],
+    schema: OpenAPIDocument,
     metadata: OIDCMetadata,
 ) -> None:
     try:
@@ -129,7 +141,7 @@ def __init__(
         self.validation_status = validation_status
         self.oidc_metadata: OIDCMetadata | None = None
 
-    def openapi(self) -> dict[str, Any]:
+    def openapi(self) -> OpenAPIDocument:
         if self.openapi_schema is not None:
             return self.openapi_schema
         schema = super().openapi()

app/io/entity.py

@@ -1,4 +1,5 @@
 from enum import StrEnum
+from typing import Annotated
 from uuid import UUID
 
 from pydantic import AwareDatetime, BaseModel, ConfigDict, Field
@@ -16,18 +17,22 @@ class OrderDirection(StrEnum):
     DESC = "desc"
 
 
+EntityName = Annotated[str, Field(min_length=1, max_length=255)]
+EntityDescription = Annotated[str | None, Field(max_length=5000)]
+
+
 class EntityCreate(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
-    name: str = Field(..., min_length=1, max_length=255)
-    description: str | None = Field(None, max_length=5000)
+    name: EntityName
+    description: EntityDescription = None
 
 
 class EntityUpdate(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
-    name: str = Field(..., min_length=1, max_length=255)
-    description: str | None = Field(None, max_length=5000)
+    name: EntityName
+    description: EntityDescription = None
 
 
 class EntityResponse(BaseModel):

app/logger/configuration.py

@@ -4,7 +4,7 @@
 
 from app.logger.handlers import LOG_LEVEL, get_logfire_handler
 
-LogHandlerFactory = Callable[[], Handler]
+type LogHandlerFactory = Callable[[], Handler]
 
 __all__ = ["LogHandlerFactory", "configure_logger", "setup_logging"]
 

app/logger/filters.py

@@ -12,7 +12,7 @@ def _record_path(record: logging.LogRecord) -> str | None:
     if isinstance(args, tuple) and len(args) > _UVICORN_ACCESS_PATH_ARG_INDEX:
         path = args[_UVICORN_ACCESS_PATH_ARG_INDEX]
         if isinstance(path, str):
-            return path.split("?", maxsplit=1)[0]
+            return path.partition("?")[0]
 
     message = record.getMessage()
     for path in HEALTH_ENDPOINT_PATHS:

app/main.py

@@ -94,7 +94,7 @@ def app_from_env() -> OIDCOpenAPIFastAPI:
     return create_app(Settings.from_env())
 
 
-def main() -> None:
+def main() -> int:
     settings = Settings.from_env()
     telemetry.configure_otel(settings.app.app_name)
     logger.setup_logging()
@@ -108,7 +108,8 @@ def main() -> None:
         proxy_headers=settings.app.proxy_headers,
         forwarded_allow_ips=settings.app.forwarded_allow_ips,
     )
+    return 0
 
 
 if __name__ == "__main__":
-    main()
+    raise SystemExit(main())

app/repository/base.py

@@ -1,16 +1,14 @@
 from collections.abc import Sequence
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, overload
+from typing import overload
+from uuid import UUID
 
 from sqlalchemy import ColumnElement, delete, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import InstrumentedAttribute
 
 from app.model.base import IntegerPrimaryKey, UUIDPrimaryKey
 
-if TYPE_CHECKING:
-    from uuid import UUID
-
 
 @dataclass(frozen=True, slots=True)
 class Ordering:
@@ -54,6 +52,7 @@ async def find_all_by_id(self, ids: Sequence[KeyType]) -> Sequence[ModelType]:
 
     async def find_all_paginated(
         self,
+        *,
         offset: int = 0,
         limit: int = 25,
         ordering: Ordering | None = None,

app/service/entity.py

@@ -24,6 +24,7 @@ async def find_by_id(self, entity_id: UUID) -> Entity | None: ...
 
     async def find_all_paginated(
         self,
+        *,
         offset: int = 0,
         limit: int = 25,
         ordering: Ordering | None = None,
@@ -50,6 +51,7 @@ async def get_by_id(self, entity_id: UUID) -> Entity:
 
     async def get_all(
         self,
+        *,
         offset: int = 0,
         limit: int = 25,
         order_by: EntityOrderBy = EntityOrderBy.CREATED_AT,

app/settings/__init__.py

@@ -1,8 +1,11 @@
+from typing import Self
+
 from pydantic import BaseModel, ConfigDict
 
 from app.settings.app import ApplicationSettings
 from app.settings.authentication import (
     AuthNSettings,
+    OIDCDiscoveryError,
     OIDCMetadata,
     resolve_oidc_metadata,
 )
@@ -17,7 +20,7 @@ class Settings(BaseModel):
     oidc: AuthNSettings
 
     @classmethod
-    def from_env(cls) -> "Settings":
+    def from_env(cls) -> Self:
         return cls(
             app=ApplicationSettings(),
             database=DatabaseSettings(),
@@ -29,6 +32,7 @@ def from_env(cls) -> "Settings":
     "ApplicationSettings",
     "AuthNSettings",
     "DatabaseSettings",
+    "OIDCDiscoveryError",
     "OIDCMetadata",
     "Settings",
     "resolve_oidc_metadata",

app/settings/authentication.py

@@ -1,4 +1,5 @@
 import asyncio
+from typing import Self
 
 import httpx
 from pydantic import BaseModel, ConfigDict, Field, model_validator
@@ -10,6 +11,10 @@
 _DISCOVERY_TIMEOUT = httpx.Timeout(5.0, connect=2.0)
 
 
+class OIDCDiscoveryError(RuntimeError):
+    pass
+
+
 class OIDCMetadata(BaseModel):
     model_config = ConfigDict(frozen=True)
 
@@ -40,7 +45,7 @@ class AuthNSettings(BaseSettings):
     token_endpoint: str | None = None
 
     @model_validator(mode="after")
-    def _validate_endpoint_overrides(self) -> "AuthNSettings":
+    def _validate_endpoint_overrides(self) -> Self:
         overrides = (
             self.jwks_uri,
             self.issuer,
@@ -77,7 +82,7 @@ def _replace_url_base(url: str, source_base: str, target_base: str) -> str:
     if url == source:
         return target
     if url.startswith(f"{source}/"):
-        return f"{target}{url[len(source) :]}"
+        return f"{target}{url.removeprefix(source)}"
     return url
 
 
@@ -145,7 +150,7 @@ async def _discover_oidc_metadata(
                 await asyncio.sleep(retry_delay_seconds * attempt)
 
     message = f"OIDC discovery failed after {attempts} attempts: {discovery_url}"
-    raise RuntimeError(message) from last_error
+    raise OIDCDiscoveryError(message) from last_error
 
 
 async def resolve_oidc_metadata(
@@ -155,7 +160,8 @@ async def resolve_oidc_metadata(
     attempts: int = _DISCOVERY_ATTEMPTS,
     retry_delay_seconds: float = _DISCOVERY_RETRY_DELAY_SECONDS,
 ) -> OIDCMetadata:
-    if override := settings.metadata_override():
+    override = settings.metadata_override()
+    if override is not None:
         return override
     if attempts < 1:
         message = "OIDC discovery attempts must be at least 1"