173 lines
4.7 KiB
Python
173 lines
4.7 KiB
Python
#import schemas.base
|
|
from typing import List, Optional
|
|
|
|
from fastapi import Depends, Request, status
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.encoders import jsonable_encoder
|
|
from fastapi.exceptions import RequestValidationError, ValidationError
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi_jwt_auth import AuthJWT
|
|
from fastapi_jwt_auth.exceptions import AuthJWTException
|
|
from fastapi_pagination import add_pagination
|
|
from sqladmin import Admin, ModelView
|
|
from sqlmodel import SQLModel, Field
|
|
from sqlmodel import Session, select
|
|
|
|
import config
|
|
import routes.base
|
|
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
|
|
from database.db import engine
|
|
from database.exercices.models import Exercice, ExerciceReadFull
|
|
from services.jwt import revoke_access, revoke_refresh
|
|
from services.password import get_password_hash
|
|
|
|
app = FastAPI(title="API Generateur d'exercices", root_path="/api/", openapi_prefix="/api/")
|
|
|
|
|
|
|
|
origins = [
|
|
"http://localhost:8000",
|
|
"https://localhost:8001",
|
|
"http://localhost",
|
|
"http://localhost:8080",
|
|
"http://localhost:5173"
|
|
]
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=['*'],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
expose_headers=['*']
|
|
)
|
|
|
|
admin = Admin(app, engine)
|
|
|
|
|
|
class UserAdmin(ModelView, model=User):
|
|
column_list = [User.id, User.username]
|
|
|
|
|
|
admin.add_view(UserAdmin)
|
|
|
|
|
|
class Id_codeField(str):
|
|
@classmethod
|
|
def __get_validators__(cls):
|
|
yield cls.validate
|
|
|
|
@classmethod
|
|
def validate(cls, value, values, config, field):
|
|
print("validator", cls, value, values, config, field)
|
|
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):
|
|
print('VAlIDATE')
|
|
session= get_session()
|
|
session = next(session)
|
|
y = session.exec(select(cls)).all()
|
|
print(y)
|
|
code = generate_unique_code(cls, s=session)
|
|
print(code)
|
|
return code '''
|
|
|
|
@app.on_event("startup")
|
|
def on_startup():
|
|
create_db_and_tables()
|
|
|
|
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}
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
@app.exception_handler(ValidationError)
|
|
async def validation_exception_handler(request: Request, exc: RequestValidationError|ValidationError):
|
|
errors = {}
|
|
print(exc.errors())
|
|
for e in exc.errors():
|
|
locs = [e for e in e['loc'] if type(e) == str]
|
|
errors[locs[-1] + "_error"] = e['msg']
|
|
|
|
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)
|
|
|
|
|
|
@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):
|
|
exercices: List[ExerciceReadFull] = None
|
|
|
|
@app.post('/test', response_model=List[ExerciceReadFull] )
|
|
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)
|