refactor: Simplify api.py (move endpoints to their own routers) (#226)

app/auth.py

@@ -1,12 +1,17 @@
-from typing import Any, Optional, cast
+from typing import Annotated, Any, Optional, cast
 
+from fastapi import Depends, status
 from fastapi.exceptions import HTTPException
 from fastapi.openapi.models import OAuthFlows as OAuthFlowsModel
 from fastapi.security import OAuth2
 from fastapi.security.utils import get_authorization_scheme_param
+from sqlalchemy.orm import Session
 from starlette.requests import Request
 from starlette.status import HTTP_401_UNAUTHORIZED
 
+from app import crud, schemas
+from app.db import get_db
+
 
 # This class is derived from FastAPI's OAuth2PasswordBearer class,
 # but adds support for cookie sessions.
@@ -51,3 +56,58 @@ async def __call__(self, request: Request) -> Optional[str]:
             else:
                 return None
         return param
+
+
+# Authentication helpers
+# ------------------------------------------------------------------------------
+oauth2_scheme = OAuth2PasswordBearerOrAuthCookie(tokenUrl="auth")
+# Version of oauth2_scheme that does not raise an error if the token is
+# invalid or missing
+oauth2_scheme_no_error = OAuth2PasswordBearerOrAuthCookie(
+    tokenUrl="auth", auto_error=False
+)
+
+
+def get_current_user(
+    token: Annotated[str, Depends(oauth2_scheme)], db: Session = Depends(get_db)
+) -> schemas.UserCreate:
+    """Get the current user if authenticated.
+
+    This function is used as a dependency in endpoints that require
+    authentication. It raises an HTTPException if the user is not
+    authenticated.
+
+    :param token: the authentication token
+    :param db: the database session
+    :raises HTTPException: if the user is not authenticated
+    :return: the current user
+    """
+    if token and "__U" in token:
+        session = crud.get_session_by_token(db, token=token)
+        if session:
+            return crud.update_session_last_used_field(db, session=session).user
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid authentication credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+
+def get_current_user_optional(
+    token: Annotated[str, Depends(oauth2_scheme_no_error)],
+    db: Session = Depends(get_db),
+) -> schemas.UserCreate | None:
+    """Get the current user if authenticated, None otherwise.
+
+    This function is used as a dependency in endpoints that require
+    authentication, but where the user is optional.
+
+    :param token: the authentication token
+    :param db: the database session
+    :return: the current user if authenticated, None otherwise
+    """
+    if token and "__U" in token:
+        session = crud.get_session_by_token(db, token=token)
+        if session:
+            return crud.update_session_last_used_field(db, session=session).user
+    return None

app/db.py

@@ -1,6 +1,6 @@
 from sqlalchemy import create_engine
 from sqlalchemy.engine import URL
-from sqlalchemy.orm import declarative_base, sessionmaker
+from sqlalchemy.orm import DeclarativeBase, sessionmaker
 
 from app.config import settings
 
@@ -17,4 +17,14 @@
 
 session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
-Base = declarative_base()
+
+class Base(DeclarativeBase):
+    pass
+
+
+def get_db():
+    db = session()
+    try:
+        yield db
+    finally:
+        db.close()

app/routers/auth.py

@@ -0,0 +1,133 @@
+import time
+import uuid
+from typing import Annotated
+
+import requests
+from fastapi import APIRouter, Depends, HTTPException, Query, Response, status
+from fastapi.security import OAuth2PasswordRequestForm
+from sqlalchemy.orm import Session
+
+from app import crud, schemas
+from app.auth import oauth2_scheme
+from app.config import settings
+from app.db import get_db
+from app.models import Session as SessionModel
+
+auth_router = APIRouter(prefix="/auth")
+session_router = APIRouter(prefix="/session")
+
+
+def create_token(user_id: str):
+    return f"{user_id}__U{str(uuid.uuid4())}"
+
+
+def get_current_session(
+    token: Annotated[str, Depends(oauth2_scheme)], db: Session = Depends(get_db)
+) -> SessionModel:
+    """Get the current user session, if authenticated.
+
+    This function is used as a dependency in endpoints that require
+    authentication. It raises an HTTPException if the user is not
+    authenticated.
+
+    :param token: the authentication token
+    :param db: the database session
+    :raises HTTPException: if the user is not authenticated
+    :return: the current user session
+    """
+    if token and "__U" in token:
+        session = crud.get_session_by_token(db, token=token)
+        if session:
+            return crud.update_session_last_used_field(db, session=session)
+    raise HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Invalid authentication credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+
+@auth_router.post("/")
+def authentication(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+    response: Response,
+    set_cookie: Annotated[
+        bool,
+        Query(
+            description="if set to 1, the token is also set as a cookie "
+            "named 'session' in the response. This parameter must be passed "
+            "as a query parameter, e.g.: /auth?set_cookie=1"
+        ),
+    ] = False,
+    db: Session = Depends(get_db),
+):
+    """
+    Authentication: provide username/password and get a bearer token in return.
+
+    - **username**: Open Food Facts user_id (not email)
+    - **password**: user password (clear text, but HTTPS encrypted)
+
+    A **token** is returned. If the **set_cookie** parameter is set to 1,
+    the token is also set as a cookie named "session" in the response.
+
+    To authenticate, you can either:
+    - use the **Authorization** header with the **Bearer** scheme,
+      e.g.: "Authorization: bearer token"
+    - use the **session** cookie, e.g.: "Cookie: session=token"
+    """
+    if "oauth2_server_url" not in settings.model_dump():
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="OAUTH2_SERVER_URL environment variable missing",
+        )
+
+    # By specifying body=1, information about the user is returned in the
+    # response, including the user_id
+    data = {"user_id": form_data.username, "password": form_data.password, "body": 1}
+    r = requests.post(f"{settings.oauth2_server_url}", data=data)  # type: ignore
+    if r.status_code == 200:
+        # form_data.username can be the user_id or the email, so we need to
+        # fetch the user_id from the response
+        # We also need to lowercase the user_id as it's case-insensitive
+        user_id = r.json()["user_id"].lower().strip()
+        token = create_token(user_id)
+        session, *_ = crud.create_session(db, user_id=user_id, token=token)
+        session = crud.update_session_last_used_field(db, session=session)
+        # set the cookie if requested
+        if set_cookie:
+            # Don't add httponly=True or secure=True as it's still in
+            # development phase, but it should be added once the front-end
+            # is ready
+            response.set_cookie(key="opsession", value=token)
+        return {"access_token": token, "token_type": "bearer"}
+    elif r.status_code == 403:
+        time.sleep(2)  # prevents brute-force
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid authentication credentials",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    raise HTTPException(
+        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Server error"
+    )
+
+
+@session_router.get("/", response_model=schemas.SessionBase)
+def get_user_session(
+    current_session: SessionModel = Depends(get_current_session),
+):
+    """Return information about the current user session."""
+    return current_session
+
+
+@session_router.delete("/")
+def delete_user_session(
+    current_session: SessionModel = Depends(get_current_session),
+    db: Session = Depends(get_db),
+):
+    """Delete the current user session.
+
+    If the provided session token or cookie is invalid, a HTTP 401 response
+    is returned.
+    """
+    crud.delete_session(db, current_session.id)
+    return {"status": "ok"}

app/routers/locations.py

@@ -0,0 +1,50 @@
+from fastapi import APIRouter, Depends, HTTPException
+from fastapi_filter import FilterDepends
+from fastapi_pagination import Page
+from fastapi_pagination.ext.sqlalchemy import paginate
+from sqlalchemy.orm import Session
+
+from app import crud, schemas
+from app.db import get_db
+
+router = APIRouter(prefix="/locations")
+
+
+@router.get("/", response_model=Page[schemas.LocationFull])
+def get_locations(
+    filters: schemas.LocationFilter = FilterDepends(schemas.LocationFilter),
+    db: Session = Depends(get_db),
+):
+    return paginate(db, crud.get_locations_query(filters=filters))
+
+
+@router.get(
+    "/osm/{location_osm_type}/{location_osm_id}",
+    response_model=schemas.LocationFull,
+)
+def get_location_by_osm(
+    location_osm_type: str, location_osm_id: int, db: Session = Depends(get_db)
+):
+    db_location = crud.get_location_by_osm_id_and_type(
+        db, osm_id=location_osm_id, osm_type=location_osm_type.upper()
+    )
+    if not db_location:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Location with type {location_osm_type} & id {location_osm_id} not found",
+        )
+    return db_location
+
+
+@router.get(
+    "/{location_id}",
+    response_model=schemas.LocationFull,
+)
+def get_location_by_id(location_id: int, db: Session = Depends(get_db)):
+    db_location = crud.get_location_by_id(db, id=location_id)
+    if not db_location:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Location with id {location_id} not found",
+        )
+    return db_location