crud api

.gitignore

@@ -1,3 +1,4 @@
 .venv
 .vscode
 __pycache__
+.env

backend_fastapi/auth.py

@@ -0,0 +1,27 @@
+from fastapi import APIRouter
+import backend_fastapi.schemas as schemas
+from backend_fastapi.repositories import add_role, add_user, get_role_all, update_role, delete_role
+from typing import List
+router = APIRouter()
+
+@router.post("/role")
+async def create_role(role: schemas.RoleCreate) -> schemas.Role:
+    return await add_role(role)
+
+@router.get("/role")
+async def get_role() -> List[schemas.Role]:
+    return await get_role_all()
+
+@router.patch("/role")
+async def change_role(role: schemas.Role, id: int) -> None:
+    return await update_role(role, id)
+
+@router.delete("/role")
+async def remove_role(id: int) -> schemas.Role:
+    return await delete_role(id)
+
+@router.post("/user")
+async def create_user(user: schemas.UserCreate) -> schemas.User:
+    import hashlib
+    user.hashed_password = hashlib.sha256(user.hashed_password.encode('utf-8')).hexdigest()
+    return await add_user(user)

backend_fastapi/database.py

@@ -0,0 +1,29 @@
+from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
+from sqlalchemy.orm import DeclarativeBase
+import os
+from dotenv import load_dotenv
+load_dotenv()
+class Model(DeclarativeBase):
+    pass
+
+async_engine = create_async_engine(
+    os.getenv("SQL_URL"),
+    connect_args={"check_same_thread": False}
+    )
+
+
+async_session = async_sessionmaker(
+        async_engine,
+        autoflush=True,
+        autocommit=False, 
+        expire_on_commit =False
+    )
+
+
+async def connect() -> None:
+    async with async_engine.begin() as conn:
+        await conn.run_sync(Model.metadata.create_all, checkfirst=True)
+
+async def disconnect() -> None:
+    if async_engine:
+        await async_engine.dispose()

backend_fastapi/models.py

@@ -0,0 +1,21 @@
+from .database import Model
+from sqlalchemy.orm import mapped_column, Mapped, relationship
+from sqlalchemy import String, Boolean, ForeignKey
+
+class Role(Model):
+    __tablename__ = "roles"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String(255),nullable=False)
+
+class User(Model):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column( primary_key=True)
+    firstname: Mapped[str] = mapped_column(String(255),nullable=False)
+    lastname: Mapped[str] = mapped_column(String(255),nullable=False)
+    email: Mapped[str] = mapped_column(String(255),nullable=False)
+    hashed_password: Mapped[str] = mapped_column(String(255),nullable=False)
+    is_active: Mapped[bool] = mapped_column(Boolean,default=True)
+    role_id: Mapped[int] = mapped_column(ForeignKey("roles.id"))
+    role: Mapped["Role"] = relationship()

backend_fastapi/repositories.py

@@ -0,0 +1,48 @@
+import backend_fastapi.models as models
+from .database import async_session
+import backend_fastapi.schemas as schemas
+from sqlalchemy import select, update, delete
+from fastapi.exceptions import HTTPException
+async def add_role(role: schemas.RoleCreate):
+    async with async_session() as session:
+        model = models.Role(name = role.name)
+        session.add(model)
+        await session.flush()
+        await session.commit()
+        return model
+    
+async def get_role_all():
+    async with async_session() as session:
+        result = await session.scalars(select(models.Role))
+        return result.all()
+        
+async def delete_role(id: int):
+    async with async_session() as session:
+        data = await session.scalars(select(models.Role).filter(models.Role.id == id))
+        result = data.one_or_none()
+        if not result:
+            raise HTTPException(status_code=404, detail="Item not found")
+        await session.execute(delete(models.Role).filter(models.Role.id == id))
+        await session.commit()
+        return result
+        
+    
+async def update_role(role: schemas.Role, id: int):
+    async with async_session() as session:
+        query = update(models.Role).filter(models.Role.id == id).values(name = role.name)
+        await session.execute(query)
+        await session.commit()
+        return {f"{id=} был обновлен"}
+
+async def add_user(user: schemas.UserCreate):
+    async with async_session() as session:
+        model = models.User(
+            **user.model_dump()
+            )
+        session.add(model)
+        await session.flush()
+        await session.commit()
+        return model
+    
+
+

backend_fastapi/schemas.py

@@ -0,0 +1,26 @@
+from pydantic import BaseModel, ConfigDict, EmailStr
+
+class RoleBase(BaseModel):
+    name: str
+
+class UserBase(BaseModel):
+    firstname: str
+    lastname: str
+    email: EmailStr
+    hashed_password: str
+    role_id: int 
+    is_active: bool = True
+    
+class RoleCreate(RoleBase):
+    pass
+
+class UserCreate(UserBase):
+    pass
+
+class User(UserBase):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+
+class Role(RoleBase):
+    model_config = ConfigDict(from_attributes=True)
+    id: int

main.py

@@ -1,7 +1,18 @@
 from fastapi import FastAPI
 from fastapi.responses import HTMLResponse
 from fastapi.middleware.cors import CORSMiddleware
-app = FastAPI()
+from contextlib import asynccontextmanager
+from backend_fastapi.auth import router as auth_router
+from backend_fastapi.database import connect, disconnect
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    await connect()
+    yield
+    await disconnect()
+
+
+app = FastAPI(lifespan=lifespan)
 origins = [
     "http://localhost",
     "http://localhost:8000",
@@ -25,7 +36,8 @@ def index():
 async def get_data():
     return {"firstname": "Котофей","lastname":"Барсикофф","age":"10"}
 
+app.include_router(router=auth_router, prefix="/auth")
 
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=8000, reload=True)
+    uvicorn.run(app, host="0.0.0.0", port=8000)

requierements.txt

@@ -2,3 +2,4 @@ uvicorn
 fastapi
 sqlalchemy
 aioodbc
+python-dotenv