Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion app/api/wps.py
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,7 @@ async def get_drawing(
500: {"model": ErrorResponse},
},
)
async def update_drawing( # noqa: PLR0913
async def update_drawing( # noqa: PLR0913, PLR0917
request: Request,
drawing_id: uuid.UUID,
admin_id: Annotated[
Expand Down
2 changes: 1 addition & 1 deletion app/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,7 @@ async def drawing_not_found_handler(_request: Request, exc: DrawingNotFoundError
@app.exception_handler(S3Error)
async def s3_error_handler(_request: Request, exc: S3Error) -> JSONResponse:
"""Handle S3 operation errors with a 500 Internal Server Error response."""
logger.exception("S3 operation failed: %s", exc.message)
logger.error("S3 operation failed: %s", exc.message)
return JSONResponse(status_code=500, content={"detail": "Storage operation failed"})


Expand Down
30 changes: 27 additions & 3 deletions app/openapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@
from functools import lru_cache
from typing import Any

from fastapi import FastAPI, Response
from fastapi import FastAPI, Response, routing
from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
from fastapi.openapi.utils import get_openapi
from fastapi.responses import HTMLResponse
Expand All @@ -24,14 +24,38 @@
_SPEC_URL = "/openapi.json"


def _iter_routes(app: FastAPI) -> list[Any]:
"""Return the app routes flattened, one entry per operation.

Since FastAPI 0.141 / Starlette 1.6, ``include_router`` stores a single lazy
``_IncludedRouter`` in ``app.routes`` instead of flattening the child routes,
so iterating ``app.routes`` directly no longer yields the individual
``APIRoute`` objects. ``iter_route_contexts`` resolves those into contexts
that expose the effective path and tags (router prefix and tags merged in),
and ``get_openapi`` accepts them in place of routes. Older FastAPI versions
allowed by our version constraint lack that helper but already flatten the
routes, so fall back to plain iteration there.
"""
iter_route_contexts = getattr(routing, "iter_route_contexts", None)
if iter_route_contexts is None: # pragma: no cover - FastAPI < 0.141
return list(app.routes)
return list(iter_route_contexts(app.routes))


def _is_internal(route: Any) -> bool:
"""Return whether a route (or route context) is tagged as internal."""
original = getattr(route, "original_route", route)
return isinstance(original, APIRoute) and INTERNAL_TAG in (getattr(route, "tags", None) or [])


def _remove_422(schema: dict[str, Any]) -> None:
for method_item in schema.get("paths", {}).values():
for param in method_item.values():
param.get("responses", {}).pop("422", None)


def _build_default_schema(app: FastAPI) -> dict[str, Any]:
routes = [r for r in app.routes if not (isinstance(r, APIRoute) and INTERNAL_TAG in r.tags)]
routes = [r for r in _iter_routes(app) if not _is_internal(r)]
tags = [t for t in (app.openapi_tags or []) if t.get("name") != INTERNAL_TAG]
schema = get_openapi(
title=app.title,
Expand All @@ -50,7 +74,7 @@ def _build_default_schema(app: FastAPI) -> dict[str, Any]:


def _build_internal_schema(app: FastAPI) -> dict[str, Any]:
routes = [r for r in app.routes if isinstance(r, APIRoute) and INTERNAL_TAG in r.tags]
routes = [r for r in _iter_routes(app) if _is_internal(r)]
tags = [t for t in (app.openapi_tags or []) if t.get("name") == INTERNAL_TAG]
schema = get_openapi(
title=f"{app.title} - Internal",
Expand Down
3 changes: 2 additions & 1 deletion app/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,8 @@ def parse_list(cls, v: str | list[str]) -> list[str]:
@lru_cache
def get_settings() -> Settings: # pragma: no cover
"""Return the cached singleton Settings instance."""
return Settings() # ty: ignore[missing-argument] for production we don't pass parameter we use environment variable
# for production we don't pass parameter, we use environment variables
return Settings()


SettingsDep = Annotated[
Expand Down
2 changes: 1 addition & 1 deletion app/tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ def settings(moto_server: str) -> Settings:
# Pydantic will automatically load any .env or .env.default file, so for testing to avoid
# any different test result between CI and local environment (in which .env file can differ)
# we make sure pydantic doesn't load the environment file with `_env_file=None`
_env_file=None, # ty:ignore[unknown-argument]
_env_file=None,
cors_origins=["http://test.com", "https://hello.com"],
cors_origin_regex=r"http://localhost:\d+",
aws_endpoint_url=moto_server,
Expand Down
2 changes: 1 addition & 1 deletion app/tests/test_drawings_endpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -271,7 +271,7 @@ def test_create_drawing_s3_failure(client: TestClient, valid_kmz_bytes: bytes, s
client=MagicMock(),
bucket=settings.aws_s3_bucket_name,
)
broken_s3.upload_drawing = AsyncMock(side_effect=S3Error("AWS error details here")) # type: ignore # noqa: PGH003
broken_s3.upload_drawing = AsyncMock(side_effect=S3Error("AWS error details here"))

broken_drawings = DrawingsService(s3=broken_s3)

Expand Down
Loading