sync

information_system_and_programming/term_5/db_management_and_automation/pytest.ini

@@ -1,4 +1,4 @@
 [pytest]
-pythonpath = . servicemanager
+pythonpath = . src
 ; env_files =
 ;     .test.env

information_system_and_programming/term_5/db_management_and_automation/servicemanager/backend/database.py

@@ -1,25 +0,0 @@
-import datetime
-from typing import Annotated
-
-from sqlalchemy import text
-from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
-from sqlalchemy.orm import DeclarativeBase, mapped_column
-
-from config import settings
-
-
-async_engine = create_async_engine(settings.DATABASE_URL_asyncpg)
-async_session_factory = async_sessionmaker(async_engine)
-
-
-intpk = Annotated[int, mapped_column(primary_key=True)]
-created_at = Annotated[datetime.datetime, mapped_column(server_default=text("TIMEZONE('utc', now())"))]
-updated_at = Annotated[datetime.datetime, mapped_column(
-        server_default=text("TIMEZONE('utc', now())"),
-        onupdate=datetime.datetime.utcnow,
-    )]
-
-
-class Base(DeclarativeBase):
-    pass
-

information_system_and_programming/term_5/db_management_and_automation/servicemanager/backend/queries/user.py

@@ -1,25 +0,0 @@
-from backend.database import Base, async_engine, async_session_factory
-
-from backend.models.user import User, Role
-
-class UserORM:
-    @staticmethod
-    async def create_tables():
-        #  Удаление и создание пустых таблиц в БД 
-        async with async_engine.begin() as conn:
-            await conn.run_sync(Base.metadata.drop_all)
-            await conn.run_sync(Base.metadata.create_all)
-
-    @staticmethod
-    async def insert_admin():
-        async with async_session_factory() as session:
-            admin = User(
-                email="admin@mail.com",
-                hashed_password="sdfasdfasdf",
-                is_active=True,
-                is_superuser=True,
-                is_verified=True,
-            )
-            session.add(admin)
-            # flush взаимодействует с БД, поэтому пишем await
-            await session.commit()

information_system_and_programming/term_5/db_management_and_automation/src/backend/database.py

@@ -0,0 +1,52 @@
+import datetime
+from typing import Annotated, AsyncGenerator
+
+from fastapi import Depends
+from fastapi_users.db import SQLAlchemyUserDatabase
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.orm import DeclarativeBase, mapped_column
+
+from config import settings
+
+async_engine = create_async_engine(settings.DATABASE_URL_asyncpg)
+async_session_maker = async_sessionmaker(
+    async_engine,
+    class_=AsyncSession,
+    expire_on_commit=False,
+)
+
+intpk = Annotated[int, mapped_column(primary_key=True)]
+created_at = Annotated[
+    datetime.datetime,
+    mapped_column(server_default=text("TIMEZONE('utc', now())")),
+]
+updated_at = Annotated[
+    datetime.datetime,
+    mapped_column(
+        server_default=text("TIMEZONE('utc', now())"),
+        onupdate=datetime.datetime.utcnow,
+    ),
+]
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+from backend.models.user import User
+
+
+async def create_db_and_tables():
+    async with async_engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+        await conn.run_sync(Base.metadata.create_all)
+
+
+async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
+    async with async_session_maker() as session:
+        yield session
+
+
+async def get_user_db(session: AsyncSession = Depends(get_async_session)):
+    yield SQLAlchemyUserDatabase(session, User)

information_system_and_programming/term_5/db_management_and_automation/src/backend/models/user.py

@@ -1,8 +1,8 @@
-from typing import Optional, List
+from typing import List, Optional
 
 from fastapi_users.db import SQLAlchemyBaseUserTable
 from sqlalchemy import ForeignKey
-from sqlalchemy.orm import Mapped, relationship, mapped_column
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from backend.database import Base, intpk
 
@@ -11,30 +11,30 @@ class UserRoleAssociation(Base):
     """
     Модель таблицы связи Many2Many для Role & User
     """
+
     __tablename__ = "user_role_association"
     user_id: Mapped[int] = mapped_column(ForeignKey("user.id"), primary_key=True)
-    role_id: Mapped[int] = mapped_column(
+    role_id: Mapped[int] = mapped_column(ForeignKey("role.id"), primary_key=True)
-        ForeignKey("role.id"), primary_key=True
-    )
     extra_data: Mapped[Optional[str]]
     child: Mapped["Role"] = relationship()
 
-    
+
 class Role(Base):
     """
     Модель таблицы role
     """
+
     __tablename__ = "role"
 
     id: Mapped[intpk]
     role: Mapped[str]
-    
+
 
 class User(SQLAlchemyBaseUserTable[int], Base):
     """
     Модель таблицы user
     """
+
     __tablename__ = "user"
     id: Mapped[intpk]
     children: Mapped[List["UserRoleAssociation"]] = relationship()
-    

information_system_and_programming/term_5/db_management_and_automation/src/backend/queries/user.py

@@ -0,0 +1,18 @@
+from backend.database import async_session_maker
+from backend.models.user import Role, User
+
+
+class UserORM:
+    @staticmethod
+    async def insert_admin():
+        async with async_session_maker() as session:
+            admin = User(
+                email="admin@mail.com",
+                hashed_password="sdfasdfasdf",
+                is_active=True,
+                is_superuser=True,
+                is_verified=True,
+            )
+            session.add(admin)
+            # flush взаимодействует с БД, поэтому пишем await
+            await session.commit()

information_system_and_programming/term_5/db_management_and_automation/src/config.py

@@ -15,7 +15,7 @@ class Settings(BaseSettings):
         """
         return f"postgresql+asyncpg://{self.DB_USER}:{self.DB_PASS}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
 
-    model_config = SettingsConfigDict(env_file=".env")
+    model_config = SettingsConfigDict(env_file="src/.env")
 
 
 settings = Settings()

information_system_and_programming/term_5/db_management_and_automation/src/frontend/routes.py

@@ -9,12 +9,12 @@ site_router = APIRouter()
 
 site_router.mount(
     "/static",
-    StaticFiles(directory="servicemanager/frontend/static"),
+    StaticFiles(directory="src/frontend/static"),
     name="static",
 )
 
 
-templates = Jinja2Templates(directory="servicemanager/frontend/templates")
+templates = Jinja2Templates(directory="src/frontend/templates")
 
 
 @site_router.get("/testpage", response_class=HTMLResponse)

information_system_and_programming/term_5/db_management_and_automation/src/main.py

@@ -2,8 +2,9 @@ import asyncio
 import sys
 
 import uvicorn
-from api.routes import api_router
 from fastapi import FastAPI
+
+from api.routes import api_router
 from frontend.routes import site_router
 
 
@@ -11,8 +12,10 @@ async def generate_test_data():
     """
     Создание БД и наполнение ее данными
     """
+    from backend.database import create_db_and_tables
     from backend.queries.user import UserORM
-    await UserORM.create_tables()
+
+    await create_db_and_tables()
     await UserORM.insert_admin()
 
 

information_system_and_programming/term_5/db_management_and_automation/tests/test_frontend.py

@@ -1,6 +1,6 @@
 from starlette.testclient import TestClient
 
-from servicemanager.main import create_app
+from src.main import create_app
 
 client = TestClient(create_app())
 

information_system_and_programming/term_5/db_management_and_automation/tests/test_main.py

@@ -1,6 +1,6 @@
 from starlette.testclient import TestClient
 
-from servicemanager.main import create_app
+from src.main import create_app
 
 client = TestClient(create_app())