generateur_v3/backend/api/main.py
Kilton937342 936931f047 api
2022-10-10 01:34:38 +02:00

175 lines
5.1 KiB
Python

#import schemas.base
from services.database import generate_unique_code
from sqlmodel import SQLModel, Field, select
from services.password import get_password_hash
from sqlmodel import Session, select
from database.auth.crud import create_user_db
from services.auth import get_current_user_optional, jwt_required
from fastapi.openapi.utils import get_openapi
from database.auth.models import User, UserRead
from database.exercices.models import Exercice, ExerciceRead
from fastapi_pagination import add_pagination
from fastapi.responses import PlainTextResponse
from fastapi.exceptions import RequestValidationError, ValidationError
from fastapi import FastAPI, HTTPException, Depends, Request, status, Header
from fastapi_jwt_auth import AuthJWT
from fastapi_jwt_auth.exceptions import AuthJWTException
from fastapi.responses import JSONResponse
from typing import List, Optional, Sequence
from tortoise.contrib.pydantic import pydantic_model_creator
from fastapi import FastAPI, HTTPException, params
from tortoise import Tortoise
from fastapi.middleware.cors import CORSMiddleware
from tortoise.contrib.fastapi import register_tortoise
from pydantic import BaseModel, validator
from database.db import create_db_and_tables, get_session
from services.jwt import revoke_access, revoke_refresh
import routes.base
from redis import Redis
from fastapi.encoders import jsonable_encoder
import config
from sqladmin import Admin, ModelView
from database.db import engine
from fastapi.security import OAuth2PasswordBearer, HTTPBearer
app = FastAPI(title="API Generateur d'exercices")
origins = [
"http://localhost:8000",
"https://localhost:8001",
"http://localhost",
"http://localhost:8080",
]
app.add_middleware(
CORSMiddleware,
allow_origins=['*'],
allow_credentials=True,
allow_methods=["*"],
allow_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, 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[ExerciceRead] = None
@app.post('/test', response_model=List[ExerciceRead] )
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)