You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
111 lines
3.9 KiB
111 lines
3.9 KiB
from fastapi import FastAPI, APIRouter, Request, HTTPException,Depends
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import ORJSONResponse, HTMLResponse
|
|
from aiohttp import ClientSession
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from sqlmodel import create_engine, Field, SQLModel, UUID, DateTime, func, Column, text, bindparam
|
|
from sqlalchemy.dialects.mssql import UNIQUEIDENTIFIER, DATETIME
|
|
from sqlalchemy import insert
|
|
from dotenv import load_dotenv
|
|
from typing import Optional, Sequence
|
|
from datetime import datetime
|
|
from uuid import UUID
|
|
import os
|
|
load_dotenv()
|
|
DB_URL = os.getenv("DB_URL")
|
|
SERVER_ROOT = os.getenv("SERVER_ROOT")
|
|
PREFIX = (SERVER_ROOT if SERVER_ROOT else "") +os.getenv("PREFIX")
|
|
URL = os.getenv("URL")
|
|
db_engine = create_engine(DB_URL)
|
|
|
|
class Agreement(SQLModel,table=True):
|
|
id: Optional[int] = Field(default=None, primary_key=True)
|
|
id_account: UUID = Field(sa_column=Column(UNIQUEIDENTIFIER))
|
|
id_personal_account: Optional[int]
|
|
date_registration:datetime = Field(sa_column=Column(DATETIME))
|
|
FIO: Optional[str]
|
|
phone: Optional[str]
|
|
email: Optional[str]
|
|
agreement:Optional[bool] = True
|
|
agreement_date: Optional[datetime] = Field(sa_column=Column(DATETIME))
|
|
|
|
class AgreementCreate(SQLModel):
|
|
id_account: UUID
|
|
id_personal_account: Optional[int]
|
|
date_registration:datetime
|
|
FIO: Optional[str]
|
|
phone: Optional[str]
|
|
email: Optional[str]
|
|
agreement:Optional[bool] = True
|
|
agreement_date: Optional[datetime]
|
|
SQLModel.metadata.create_all(db_engine, checkfirst=True)
|
|
|
|
app = FastAPI(openapi_url = SERVER_ROOT +"/openapi.json", docs_url=SERVER_ROOT+"/docs", default_response_class=ORJSONResponse)
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
async def get_current_user(data: str = Depends(OAuth2PasswordBearer(tokenUrl=f"{URL}/auth/login/"))):
|
|
import json
|
|
async with ClientSession() as session:
|
|
response = await session.get(f"{URL}/auth/get_current_user/{data}")
|
|
data = await response.read()
|
|
return json.loads(data)
|
|
|
|
router = APIRouter(
|
|
prefix=PREFIX,
|
|
dependencies=[Depends(get_current_user)]
|
|
)
|
|
|
|
@router.get('/')
|
|
async def get_ageements(id_account: Optional[UUID] = None) -> Sequence[Agreement]:
|
|
with db_engine.begin() as con:
|
|
if id_account:
|
|
query = text("SELECT * FROM agreement WHERE id_account=:id_account").params({"id_account":id_account})
|
|
else:
|
|
query = text("SELECT * FROM agreement")
|
|
return con.execute(query).fetchall()
|
|
|
|
|
|
@router.delete("/")
|
|
async def delete_agreements(acc_list: Sequence[UUID]):
|
|
if acc_list:
|
|
with db_engine.begin() as con:
|
|
stmt = text("DELETE agreement WHERE id_account IN :acc_list")
|
|
params = {"acc_list":acc_list}
|
|
stmt = stmt.bindparams(bindparam('acc_list', expanding=True))
|
|
con.execute(stmt, params)
|
|
return {"msg":"has deleted"}
|
|
raise HTTPException(405,"list is empty")
|
|
|
|
@router.post("/")
|
|
async def add_agreements(agrees: Sequence[AgreementCreate]):
|
|
if agrees:
|
|
with db_engine.begin() as con:
|
|
dict_list=[]
|
|
acc_list = []
|
|
for agree in agrees:
|
|
agree_dict = agree.model_dump()
|
|
dict_list.append(agree_dict)
|
|
acc_id = agree_dict.get("id_account")
|
|
if acc_id not in acc_list:
|
|
acc_list.append(acc_id)
|
|
await delete_agreements(acc_list)
|
|
stmt = insert(Agreement).values(dict_list)
|
|
con.execute(stmt)
|
|
con.commit()
|
|
return {"msg": "added succesfully"}
|
|
raise HTTPException(405,"list is empty")
|
|
app.include_router(router)
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(
|
|
app,
|
|
host=os.getenv("HOST"),
|
|
port=int(os.getenv("PORT")),
|
|
)
|