Anaz
/
After_Chido_Api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

factorisation db.py, database.py

main
Anaz 2025-01-10 10:00:13 +04:00
parent 1a7d66b09d
commit e3732e1c9a
2 changed files with 38 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
config/database.py
View File

@ -1,129 +1,17 @@
from datetime import datetime
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import sessionmaker
from sqlalchemy import Table, Column, Integer, String, DateTime, Boolean, MetaData, ForeignKey, Text
from config.settings import settings from config.settings import settings
import logging import logging
# Configuration du logging
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Métadonnées pour les tables
metadata = MetaData()
# Table des utilisateurs
users_table = Table(
"users",
metadata,
Column("id", Integer, primary_key=True),
Column("email", String(255), unique=True, nullable=False), # VARCHAR(255)
Column("full_name", String(255), nullable=False), # VARCHAR(255)
Column("phone", String(20), nullable=False), # VARCHAR(20)
Column("date_of_birth", DateTime, nullable=False),
Column("organization", String(255)), # VARCHAR(255)
Column("hashed_password", String(255), nullable=False), # VARCHAR(255)
Column("role", String(50), default="user"), # VARCHAR(50)
Column("is_blocked", Boolean, default=False),
Column("is_deleted", Boolean, default=False)
)
# Table des rôles
roles_table = Table(
"roles",
metadata,
Column("id", Integer, primary_key=True),
Column("name", String(50), unique=True, nullable=False), # VARCHAR(50)
Column("permissions", Text) # Stocke les permissions sous forme de chaîne séparée par des virgules
)
# Table des demandes de besoin
need_requests_table = Table(
"need_requests",
metadata,
Column("id", Integer, primary_key=True),
Column("category", String(50), nullable=False), # Ex: "water", "food", "medical assistance"
Column("description", Text, nullable=False),
Column("adults", Integer, nullable=False),
Column("children", Integer, nullable=False),
Column("vulnerable", Integer, nullable=False),
Column("location", String(255), nullable=False), # VARCHAR(255)
Column("gps_coordinates", String(100)), # VARCHAR(100)
Column("requester_email", String(255), ForeignKey("users.email"), nullable=False), # VARCHAR(255)
Column("assigned_to", Integer, ForeignKey("users.id")), # Utilisateur qui a pris en charge la demande
Column("status", String(50), default="pending") # Ex: "pending", "in_progress", "completed"
)
# Table des signalements d'utilisateurs
user_reports_table = Table(
"user_reports",
metadata,
Column("id", Integer, primary_key=True),
Column("reporter_id", Integer, ForeignKey("users.id"), nullable=False),
Column("reported_user_id", Integer, ForeignKey("users.id"), nullable=False),
Column("reason", Text, nullable=False),
Column("status", String(50), default="pending") # Ex: "pending", "resolved"
)
# Table des problèmes techniques
technical_issues_table = Table(
"technical_issues",
metadata,
Column("id", Integer, primary_key=True),
Column("user_id", Integer, ForeignKey("users.id"), nullable=False),
Column("description", Text, nullable=False),
Column("status", String(50), default="open") # Ex: "open", "in_progress", "resolved"
)
# Table des points d'intérêt
points_of_interest_table = Table(
"points_of_interest",
metadata,
Column("id", Integer, primary_key=True),
Column("label", String(255), nullable=False), # VARCHAR(255)
Column("description", Text),
Column("icon", String(255)), # URL de l'icône (VARCHAR(255))
Column("organization", String(255)), # VARCHAR(255)
Column("gps_coordinates", String(100), nullable=False), # VARCHAR(100)
Column("added_by", Integer, ForeignKey("users.id"), nullable=False)
)
# Table des abris
shelters_table = Table(
"shelters",
metadata,
Column("id", Integer, primary_key=True),
Column("label", String(255), nullable=False), # VARCHAR(255)
Column("description", Text),
Column("status", String(50), default="available"), # Ex: "available", "full", "closed"
Column("contact_person", String(255)), # VARCHAR(255)
Column("gps_coordinates", String(100), nullable=False), # VARCHAR(100)
Column("added_by", Integer, ForeignKey("users.id"), nullable=False)
)
# Table des déclarations de personnes
person_reports_table = Table(
"person_reports",
metadata,
Column("id", Integer, primary_key=True),
Column("full_name", String(255), nullable=False), # VARCHAR(255)
Column("date_of_birth", DateTime, nullable=False),
Column("status", String(50), nullable=False), # Ex: "missing", "safe", "deceased"
Column("location", String(255)), # VARCHAR(255)
Column("gps_coordinates", String(100)), # VARCHAR(100)
Column("photo_url", String(255)), # URL de la photo (VARCHAR(255))
Column("reporter_email", String(255), ForeignKey("users.email"), nullable=False), # VARCHAR(255)
Column("created_at", DateTime, default=datetime.utcnow), # Date de création
Column("updated_at", DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) # Date de mise à jour
)
# Connexion à la base de données
DATABASE_URL = settings.database_url DATABASE_URL = settings.database_url
engine = create_async_engine(DATABASE_URL, echo=True) engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
# Fonction pour initialiser la base de données
async def init_db(): async def init_db():
from models.db import metadata
try: try:
async with engine.begin() as conn: async with engine.begin() as conn:
await conn.run_sync(metadata.create_all) await conn.run_sync(metadata.create_all)
@ -132,7 +20,6 @@ async def init_db():
logger.error(f"Error creating database tables: {str(e)}") logger.error(f"Error creating database tables: {str(e)}")
raise raise
# Fonction pour obtenir une session de base de données
async def get_db(): async def get_db():
async with AsyncSessionLocal() as session: async with AsyncSessionLocal() as session:
yield session yield session

108
models/db.py
View File

@ -1,61 +1,49 @@
from datetime import datetime
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Table, Column, Integer, String, DateTime, Boolean, MetaData, ForeignKey, Text from sqlalchemy import Table, Column, Integer, String, DateTime, Boolean, MetaData, ForeignKey, Text
from config.settings import settings from datetime import datetime
import logging
# Configuration du logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Métadonnées pour les tables
metadata = MetaData() metadata = MetaData()
# Table des utilisateurs # Table definitions
users_table = Table( users_table = Table(
"users", "users",
metadata, metadata,
Column("id", Integer, primary_key=True), Column("id", Integer, primary_key=True),
Column("email", String(255), unique=True, nullable=False), # VARCHAR(255) Column("email", String(255), unique=True, nullable=False),
Column("full_name", String(255), nullable=False), # VARCHAR(255) Column("full_name", String(255), nullable=False),
Column("phone", String(20), nullable=False), # VARCHAR(20) Column("phone", String(20), nullable=False),
Column("date_of_birth", DateTime, nullable=False), Column("date_of_birth", DateTime, nullable=False),
Column("organization", String(255)), # VARCHAR(255) Column("organization", String(255)),
Column("hashed_password", String(255), nullable=False), # VARCHAR(255) Column("hashed_password", String(255), nullable=False),
Column("role", String(50), default="user"), # VARCHAR(50) Column("role", String(50), default="user"),
Column("is_blocked", Boolean, default=False), Column("is_blocked", Boolean, default=False),
Column("is_deleted", Boolean, default=False) Column("is_deleted", Boolean, default=False)
) )
# Table des rôles
roles_table = Table( roles_table = Table(
"roles", "roles",
metadata, metadata,
Column("id", Integer, primary_key=True), Column("id", Integer, primary_key=True),
Column("name", String(50), unique=True, nullable=False), # VARCHAR(50) Column("name", String(50), unique=True, nullable=False),
Column("permissions", Text) # Stocke les permissions sous forme de chaîne séparée par des virgules Column("permissions", Text)
) )
# Table des demandes de besoin
need_requests_table = Table( need_requests_table = Table(
"need_requests", "need_requests",
metadata, metadata,
Column("id", Integer, primary_key=True), Column("id", Integer, primary_key=True),
Column("category", String(50), nullable=False), # Ex: "water", "food", "medical assistance" Column("category", String(50), nullable=False),
Column("description", Text, nullable=False), Column("description", Text, nullable=False),
Column("adults", Integer, nullable=False), Column("adults", Integer, nullable=False),
Column("children", Integer, nullable=False), Column("children", Integer, nullable=False),
Column("vulnerable", Integer, nullable=False), Column("vulnerable", Integer, nullable=False),
Column("location", String(255), nullable=False), # VARCHAR(255) Column("location", String(255), nullable=False),
Column("gps_coordinates", String(100)), # VARCHAR(100) Column("gps_coordinates", String(100)),
Column("requester_email", String(255), ForeignKey("users.email"), nullable=False), # VARCHAR(255) Column("requester_email", String(255), ForeignKey("users.email"), nullable=False),
Column("assigned_to", Integer, ForeignKey("users.id")), # Utilisateur qui a pris en charge la demande Column("assigned_to", Integer, ForeignKey("users.id")),
Column("status", String(50), default="pending"), # Ex: "pending", "in_progress", "completed" Column("status", String(50), default="pending"),
Column("deleted", DateTime, default=None) # Marque la suppression logique Column("deleted", DateTime, default=None)
) )
# Table des signalements d'utilisateurs
user_reports_table = Table( user_reports_table = Table(
"user_reports", "user_reports",
metadata, metadata,
@ -63,77 +51,53 @@ user_reports_table = Table(
Column("reporter_id", Integer, ForeignKey("users.id"), nullable=False), Column("reporter_id", Integer, ForeignKey("users.id"), nullable=False),
Column("reported_user_id", Integer, ForeignKey("users.id"), nullable=False), Column("reported_user_id", Integer, ForeignKey("users.id"), nullable=False),
Column("reason", Text, nullable=False), Column("reason", Text, nullable=False),
Column("status", String(50), default="pending") # Ex: "pending", "resolved" Column("status", String(50), default="pending")
) )
# Table des problèmes techniques
technical_issues_table = Table( technical_issues_table = Table(
"technical_issues", "technical_issues",
metadata, metadata,
Column("id", Integer, primary_key=True), Column("id", Integer, primary_key=True),
Column("user_id", Integer, ForeignKey("users.id"), nullable=False), Column("user_id", Integer, ForeignKey("users.id"), nullable=False),
Column("description", Text, nullable=False), Column("description", Text, nullable=False),
Column("status", String(50), default="open") # Ex: "open", "in_progress", "resolved" Column("status", String(50), default="open")
) )
# Table des points d'intérêt
points_of_interest_table = Table( points_of_interest_table = Table(
"points_of_interest", "points_of_interest",
metadata, metadata,
Column("id", Integer, primary_key=True), Column("id", Integer, primary_key=True),
Column("label", String(255), nullable=False), # VARCHAR(255) Column("label", String(255), nullable=False),
Column("description", Text), Column("description", Text),
Column("icon", String(255)), # URL de l'icône (VARCHAR(255)) Column("icon", String(255)),
Column("organization", String(255)), # VARCHAR(255) Column("organization", String(255)),
Column("gps_coordinates", String(100), nullable=False), # VARCHAR(100) Column("gps_coordinates", String(100), nullable=False),
Column("added_by", Integer, ForeignKey("users.id"), nullable=False) Column("added_by", Integer, ForeignKey("users.id"), nullable=False)
) )
# Table des abris
shelters_table = Table( shelters_table = Table(
"shelters", "shelters",
metadata, metadata,
Column("id", Integer, primary_key=True), Column("id", Integer, primary_key=True),
Column("label", String(255), nullable=False), # VARCHAR(255) Column("label", String(255), nullable=False),
Column("description", Text), Column("description", Text),
Column("status", String(50), default="available"), # Ex: "available", "full", "closed" Column("status", String(50), default="available"),
Column("contact_person", String(255)), # VARCHAR(255) Column("contact_person", String(255)),
Column("gps_coordinates", String(100), nullable=False), # VARCHAR(100) Column("gps_coordinates", String(100), nullable=False),
Column("added_by", Integer, ForeignKey("users.id"), nullable=False) Column("added_by", Integer, ForeignKey("users.id"), nullable=False)
) )
# Table des déclarations de personnes
person_reports_table = Table( person_reports_table = Table(
"person_reports", "person_reports",
metadata, metadata,
Column("id", Integer, primary_key=True), Column("id", Integer, primary_key=True),
Column("full_name", String(255), nullable=False), # VARCHAR(255) Column("full_name", String(255), nullable=False),
Column("date_of_birth", DateTime, nullable=False), Column("date_of_birth", DateTime, nullable=False),
Column("status", String(50), nullable=False), # Ex: "missing", "safe", "deceased" Column("status", String(50), nullable=False),
Column("location", String(255)), # VARCHAR(255) Column("location", String(255)),
Column("gps_coordinates", String(100)), # VARCHAR(100) Column("gps_coordinates", String(100)),
Column("photo_url", String(255)), # URL de la photo (VARCHAR(255)) Column("photo_url", String(255)),
Column("reporter_email", String(255), ForeignKey("users.email"), nullable=False), # VARCHAR(255) Column("reporter_email", String(255), ForeignKey("users.email"), nullable=False),
Column("created_at", DateTime, default=datetime.utcnow), # Date de création Column("created_at", DateTime, default=datetime.utcnow),
Column("updated_at", DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) # Date de mise à jour Column("updated_at", DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
) )
# Connexion à la base de données
DATABASE_URL = settings.database_url
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
# Fonction pour initialiser la base de données
async def init_db():
try:
async with engine.begin() as conn:
await conn.run_sync(metadata.create_all)
logger.info("Database tables created successfully.")
except Exception as e:
logger.error(f"Error creating database tables: {str(e)}")
raise
# Fonction pour obtenir une session de base de données
async def get_db():
async with AsyncSessionLocal() as session:
yield session