from fastapi import FastAPI, APIRouter, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import ORJSONResponse, HTMLResponse
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 = os.getenv("PREFIX")
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=["*"],
)
router = APIRouter(
prefix=PREFIX
)
@app.get("/", response_class= HTMLResponse)
async def hello(request: Request):
return """
Привет
Это API, а не сайт!
"""
@router.get('/')
async def get_ageement(id_account: Optional[UUID] = None) -> Sequence[Agreement]:
with db_engine.begin() as con:
query = text("SELECT * FROM agreement WHERE id_account=:id_account").params({"id_account":id_account})
return con.execute(query).fetchall()
@router.delete("/")
async def delete_pk(acc_list: Sequence[UUID]):
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"}
@router.post("/")
async def add_agreement(agrees: Sequence[AgreementCreate]):
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_pk(acc_list)
stmt = insert(Agreement).values(dict_list)
con.execute(stmt)
con.commit()
return {"msg": "added succesfully"}
app.include_router(router)
if __name__ == "__main__":
import uvicorn
uvicorn.run(
app,
host=os.getenv("HOST"),
port=int(os.getenv("PORT"))
)