1  """
  2  ## Authentication
  3  - To authenticate requests, the `Authorization` header must contain a valid access token (JWT which contains the user's
  4    ID and the session ID).
  5  - The access token can be obtained by logging in to an exising account (see `POST /sessions` and `POST /sessions/oauth`)
  6    or by creating an account (see `POST /users`). This access token is only valid for a short period of time
  7    (usually 5 minutes).
  8  - If the access token is expired, a new access token can be obtained by using the refresh token (see `PUT /session`)
  9    which is also returned when creating a session. This will also invalidate the refresh token and generate a new one.
 10  - If the refresh token is not used to refresh the session within a configured period of time (usually 30 days) the
 11    session expires and the user must log in again on this device.
 12  
 13  ## Special parameters
 14  - In addition to the usual user ids the `user_id` path parameter used in most endpoints also accepts the special values
 15    `me` and `self` which refer to the currently authenticated user.
 16  
 17  ## Requirements
 18  Some endpoints require one or more of the following conditions to be met:
 19  - **USER**: The user is authenticated and has a valid session.
 20  - **SELF**: The authenticated user must be the same as the affected user. Requires **USER**.
 21  - **ADMIN**: The authenticated user must be an admin. Requires **USER**.
 22  - **AUTH**: The request is authenticated using a valid API token (static/JWT).
 23  """
 24  
 25  import asyncio
 26  from typing import Awaitable, Callable, TypeVar
 27  
 28  from fastapi import FastAPI, HTTPException, Request
 29  from fastapi.exception_handlers import http_exception_handler
 30  from fastapi.middleware.cors import CORSMiddleware
 31  from fastapi.responses import Response
 32  from starlette.exceptions import HTTPException as StarletteHTTPException
 33  
 34  from .database import db, db_context
 35  from .endpoints import ROUTER, TAGS
 36  from .logger import get_logger, setup_sentry
 37  from .models import User
 38  from .models.session import clean_expired_sessions
 39  from .settings import settings
 40  from .utils.debug import check_responses
 41  from .utils.docs import add_endpoint_links_to_openapi_docs
 42  from .version import get_version
 43  
 44  
 45  T = TypeVar("T")
 46  
 47  logger = get_logger(__name__)
 48  
 49  app = FastAPI(
 50      title="FastAPI",
 51      description=__doc__,
 52      version=get_version().description,
 53      root_path=settings.root_path,
 54      root_path_in_servers=False,
 55      servers=[{"url": settings.root_path}] if settings.root_path else None,
 56      openapi_tags=TAGS,
 57  )
 58  app.include_router(ROUTER)
 59  
 60  if settings.debug:
 61      app.middleware("http")(check_responses)
 62  
 63  
 64  def setup_app() -> None:
 65      add_endpoint_links_to_openapi_docs(app.openapi())
 66  
 67      if settings.sentry_dsn:
 68          logger.debug("initializing sentry")
 69          setup_sentry(app, settings.sentry_dsn, "FastAPI", get_version().description)
 70  
 71      if settings.debug:
 72          app.add_middleware(
 73              CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"]
 74          )
 75  
 76  
 77  @app.middleware("http")
 78  async def db_session(request: Request, call_next: Callable[..., Awaitable[T]]) -> T:
 79      async with db_context():
 80          return await call_next(request)
 81  
 82  
 83  @app.exception_handler(StarletteHTTPException)
 84  async def rollback_on_exception(request: Request, exc: HTTPException) -> Response:
 85      await db.session.rollback()
 86      return await http_exception_handler(request, exc)
 87  
 88  
 89  async def clean_expired_sessions_loop() -> None:
 90      while True:
 91          try:
 92              await clean_expired_sessions()
 93          except Exception as e:
 94              logger.exception(e)
 95          await asyncio.sleep(20 * 60)
 96  
 97  
 98  @app.on_event("startup")
 99  async def on_startup() -> None:
100      setup_app()
101  
102      asyncio.create_task(clean_expired_sessions_loop())
103  
104      async with db_context():
105          await User.initialize()
106  
107  
108  @app.on_event("shutdown")
109  async def on_shutdown() -> None:
110      pass
111  
112  
113  @app.head("/status", include_in_schema=False)
114  async def status() -> None:
115      pass