app/main.py - master - durovcode/backend

app/main.py
113 строк3 KB
quyxishi
implemented wishes feature for ai/query method
22 сен 2024, 06:21
22 сен 2024, 06:21bffb994
from typing import List

from fastapi import FastAPI, Path, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from fastapi_csrf_protect.exceptions import CsrfProtectError
from fastapi_redis_cache import FastApiRedisCache
from fastapi.staticfiles import StaticFiles
from sqladmin import Admin, ModelView
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.exceptions import HTTPException

from app.api.main import api_router
from app.database import alchemy_engine
from app.env import REDIS_DB_URI
from app.models import ErrorModel, Users


# RedisCache : Initialize


redis_cache = FastApiRedisCache()

redis_cache.init(
    host_url=REDIS_DB_URI,
    prefix='durovcode-cache',
    response_header='DurovCode-Cache',
    ignore_arg_types=[Request, Response, AsyncSession],
)


# FastAPI : Initialize


app = FastAPI(title='DurovCodeHackathon')
app.include_router(router=api_router, prefix='/api/v1')


app.add_middleware(
    middleware_class=CORSMiddleware,
    allow_origins=['http://localhost:3000'],
    allow_credentials=True,
    allow_methods=['*'],
    allow_headers=['*'],
)


app.mount(path='/static', app=StaticFiles(directory='static'), name='static')


# SqlAdmin : Initialize


admin = Admin(app=app, engine=alchemy_engine)


class AdminViewUsers(ModelView, model=Users):
    name_plural: str = 'Users'

    column_list: List[str] = [
        Users.id,
        Users.email,
        Users.name,
        Users.passwd_hash,
        Users.salt,
        Users.otp_key,
        Users.profile_photo,
    ]


admin.add_view(view=AdminViewUsers)


# FastAPI : Exception handlers


@app.exception_handler(exc_class_or_status_code=HTTPException)
async def http_error_handler(_, cls: HTTPException) -> FileResponse:
    return JSONResponse(
        content=ErrorModel(detail=cls.detail).model_dump(),
        status_code=cls.status_code,
        headers=cls.headers,
    )


@app.exception_handler(exc_class_or_status_code=CsrfProtectError)
async def csrf_error_handler(_, cls: CsrfProtectError) -> JSONResponse:
    return JSONResponse(
        content=ErrorModel(detail=cls.message).model_dump(),
        status_code=400,
    )


@app.exception_handler(exc_class_or_status_code=RequestValidationError)
async def validation_error_handler(_, cls: RequestValidationError) -> JSONResponse:
    return JSONResponse(
        content=ErrorModel(
            detail='; '.join(filter(None, [error.get('msg') for error in cls.errors()]))
        ).model_dump(),
        status_code=422,
    )


@app.exception_handler(exc_class_or_status_code=Exception)
async def unexpected_error_handler(_, cls: Exception) -> JSONResponse:
    return JSONResponse(
        content=ErrorModel(
            detail='Unexpected error, please contact developer about this issue and try again later'
            + f' <{(lambda x: Path(default=x.parent()).joinpath(x.name()))(Path(cls.__traceback__.tb_frame.f_code.co_filename))}:{cls.__traceback__.tb_lineno}:{cls.__class__!s}>'
        ).model_dump(),
        status_code=500,
    )