generateur_v3/backend/api/main.py

173 lines
4.7 KiB
Python
Raw Normal View History

2022-09-16 21:50:55 +02:00
#import schemas.base
2023-02-22 12:43:39 +01:00
from typing import List, Optional
from fastapi import Depends, Request, status
from fastapi import FastAPI, HTTPException
from fastapi.encoders import jsonable_encoder
2022-09-16 21:50:55 +02:00
from fastapi.exceptions import RequestValidationError, ValidationError
2023-02-22 12:43:39 +01:00
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
2022-09-16 21:50:55 +02:00
from fastapi_jwt_auth import AuthJWT
from fastapi_jwt_auth.exceptions import AuthJWTException
2023-02-22 12:43:39 +01:00
from fastapi_pagination import add_pagination
2022-09-16 21:50:55 +02:00
from sqladmin import Admin, ModelView
2023-02-22 12:43:39 +01:00
from sqlmodel import SQLModel, Field
from sqlmodel import Session, select
import config
import routes.base
2023-02-28 11:22:14 +01:00
import routes.room.routes
2023-02-22 12:43:39 +01:00
from database.auth.crud import create_user_db
from database.auth.models import User, UserRead
from database.db import create_db_and_tables, get_session
2022-09-16 21:50:55 +02:00
from database.db import engine
2023-02-22 12:43:39 +01:00
from database.exercices.models import Exercice, ExerciceReadFull
from services.jwt import revoke_access, revoke_refresh
from services.password import get_password_hash
2023-02-23 17:29:10 +01:00
app = FastAPI(title="API Generateur d'exercices", root_path="/api/", openapi_prefix="/api/")
2022-09-16 21:50:55 +02:00
origins = [
"http://localhost:8000",
"https://localhost:8001",
"http://localhost",
"http://localhost:8080",
2023-01-27 21:41:08 +01:00
"http://localhost:5173"
2022-09-16 21:50:55 +02:00
]
app.add_middleware(
CORSMiddleware,
allow_origins=['*'],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
2023-01-27 21:41:08 +01:00
expose_headers=['*']
2022-09-16 21:50:55 +02:00
)
admin = Admin(app, engine)
class UserAdmin(ModelView, model=User):
column_list = [User.id, User.username]
2022-09-26 10:04:02 +02:00
2022-10-10 01:34:38 +02:00
2022-09-16 21:50:55 +02:00
admin.add_view(UserAdmin)
2022-10-10 01:34:38 +02:00
class Id_codeField(str):
@classmethod
def __get_validators__(cls):
yield cls.validate
@classmethod
def validate(cls, value, values, config, field):
2023-02-28 10:21:08 +01:00
2022-10-10 01:34:38 +02:00
return value
class Test(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
id_code: Id_codeField
''' @validator('id_code', always=True)
def test(cls,value, values):
2023-02-28 10:21:08 +01:00
2022-10-10 01:34:38 +02:00
session= get_session()
session = next(session)
y = session.exec(select(cls)).all()
2023-02-28 10:21:08 +01:00
2022-10-10 01:34:38 +02:00
code = generate_unique_code(cls, s=session)
2023-02-28 10:21:08 +01:00
2022-10-10 01:34:38 +02:00
return code '''
2022-09-16 21:50:55 +02:00
@app.on_event("startup")
def on_startup():
create_db_and_tables()
2022-10-10 01:34:38 +02:00
def t(test_1: str):
return test_1 + "lol"
def t2(test_1: str, test_2: str):
return test_1 + test_2
def t3(test_1: str):
if test_1 == '3':
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="non")
@app.post('/test/{test_1}/{test_2}', dependencies=[Depends(t3)])
def test(test_1: str, test_2: str, test_3: str = Depends(t), test_4: str = Depends(t2)):
return {"t1": test_1, "t2": test_2, "t3": test_3, "t4": test_4}
2022-09-16 21:50:55 +02:00
@app.exception_handler(RequestValidationError)
@app.exception_handler(ValidationError)
2023-02-22 12:43:39 +01:00
async def validation_exception_handler(request: Request, exc: RequestValidationError|ValidationError):
2022-09-16 21:50:55 +02:00
errors = {}
2023-02-28 10:21:08 +01:00
2022-09-16 21:50:55 +02:00
for e in exc.errors():
2022-10-10 01:34:38 +02:00
locs = [e for e in e['loc'] if type(e) == str]
errors[locs[-1] + "_error"] = e['msg']
2022-09-16 21:50:55 +02:00
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content=jsonable_encoder({"detail": errors}),
)
#JWT AUTH
@AuthJWT.load_config
def get_config():
return config.settings
# exception handler for authjwt
# in production, you can tweak performance using orjson response
@app.exception_handler(AuthJWTException)
def authjwt_exception_handler(request: Request, exc: AuthJWTException):
return JSONResponse(
status_code=exc.status_code,
content={"detail": exc.message}
)
#REDIS
'''
@AuthJWT.token_in_denylist_loader
def check_if_token_in_denylist(decrypted_token):
jti = decrypted_token['jti']
entry = config.redis_conn.get(jti)
return entry and entry == 'true'
'''
#ROUTES
app.include_router(routes.base.api_router)
2023-02-28 11:22:14 +01:00
app.include_router(routes.room.routes.ws_router)
2022-09-16 21:50:55 +02:00
@app.delete('/access-revoke')
def access_revoke(Authorize: AuthJWT = Depends()):
Authorize.jwt_required()
revoke_access(Authorize.get_raw_jwt())
return {"detail": "Access token has been revoke"}
@app.delete('/refresh-revoke')
def refresh_revoke(Authorize: AuthJWT = Depends()):
Authorize.jwt_refresh_token_required()
revoke_refresh(Authorize.get_raw_jwt())
return {"detail": "Refresh token has been revoke"}
class user(UserRead):
2022-12-27 18:26:05 +01:00
exercices: List[ExerciceReadFull] = None
2022-09-16 21:50:55 +02:00
2022-12-27 18:26:05 +01:00
@app.post('/test', response_model=List[ExerciceReadFull] )
2022-09-16 21:50:55 +02:00
def test(db:Session= Depends(get_session)):
#create_user_db('lilian', get_password_hash('Pomme937342'), db)
create_user_db('lilian2', get_password_hash('Pomme937342'), db)
exos = db.exec(select(Exercice)).all()
return exos
add_pagination(app)