FastAPI_Base_Project

1
from contextlib import asynccontextmanager
2

3
from fastapi import FastAPI, APIRouter
4
from fastapi.middleware.gzip import GZipMiddleware
5
from fastapi_cache import FastAPICache
6
from fastapi_cache.backends.redis import RedisBackend
7
from redis import asyncio as aioredis
8

9
# from src.api.v1 import doc_router, user_router
10
from src.core import settings
11
from src.core.middleware.process_time import ProcessTimeMiddleware
12
from src.core.middleware.token_pass import PassByTokenMiddleware
13
from src.core.settings import ACCESS_TYPE
14

15
# add group router
16

17
app = FastAPI()
18

19

20
# ping-pong router
21
@app.get("/ping", tags=["Ping-Pong", ACCESS_TYPE.neutral])
22
async def ping_pong():
23
    return "pong"
24

25

26
# Adding routers
27
routers: list[APIRouter] = [
28
    # doc_router,
29
    # user_router,
30
]
31
for router in routers:
32
    app.include_router(router)
33

34

35
# Crete the dict of token_type_pass - uri
36
pass_uri = {
37
    settings.ACCESS_TYPE.access: (),
38
    settings.ACCESS_TYPE.neutral: (),
39
    settings.ACCESS_TYPE.refresh: (),
40
}
41
for route in app.router.__dict__["routes"]:
42
    if hasattr(route, "tags"):
43
        pass_uri[route.__dict__["tags"][-1]] += (route.__dict__["path"],)
44

45

46
# Middlewares
47
app.add_middleware(ProcessTimeMiddleware)
48
app.add_middleware(PassByTokenMiddleware, pass_uri=pass_uri)
49
app.add_middleware(GZipMiddleware)
50

51

52
# Adding caching
53
@asynccontextmanager
54
async def lifespan():
55
    redis = aioredis.from_url(settings.redis.REDIS_URL, encoding="utf8", decode_responses=True)
56
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")
57
    yield
58