Set up database configuration and connection management

- Created app/database.py with SQLAlchemy engine, session management, and connection utilities
- Added comprehensive database models in app/models.py for User, Client, Phone, Case, Transaction, Document, and Payment
- Implemented FastAPI application with database lifecycle management in app/main.py
- Added health check endpoint to verify database connectivity
- Created README.md with database configuration documentation
- Verified database connection works correctly with SQLite backend
This commit is contained in:
HotSwapp
2025-10-06 18:22:18 -05:00
parent 36dffd5372
commit de983a73d2
8 changed files with 457 additions and 2 deletions

91
app/database.py Normal file
View File

@@ -0,0 +1,91 @@
"""
Database configuration and connection management for Delphi Database application.
This module handles SQLAlchemy engine creation, session management, and provides
database connection utilities for the FastAPI application.
"""
import os
from typing import Generator
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
# Database configuration from environment variables
DATABASE_URL = os.getenv(
"DATABASE_URL",
"sqlite:///./delphi.db" # Default to SQLite for development
)
# Create SQLAlchemy engine
engine = create_engine(
DATABASE_URL,
connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {},
pool_pre_ping=True, # Verify connections before reuse
echo=False # Set to True for SQL query logging in development
)
# Create session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Create declarative base for models
Base = declarative_base()
def get_db() -> Generator[Session, None, None]:
"""
Dependency function that provides a database session.
Yields a database session and ensures it's properly closed after use.
Used as a FastAPI dependency in route handlers.
Yields:
Session: SQLAlchemy database session
Example:
@app.get("/items/")
async def read_items(db: Session = Depends(get_db)):
return db.query(Item).all()
"""
db = SessionLocal()
try:
yield db
finally:
db.close()
def create_tables() -> None:
"""
Create all database tables defined in SQLAlchemy models.
This function should be called during application startup to ensure
all tables exist in the database.
"""
Base.metadata.create_all(bind=engine)
def get_database_url() -> str:
"""
Get the current database URL (with sensitive info masked).
Returns:
str: Database URL with password masked for logging
"""
if "sqlite" in DATABASE_URL:
return DATABASE_URL
# For PostgreSQL/MySQL, mask the password
if "@" in DATABASE_URL:
parts = DATABASE_URL.split("@")
if "://" in parts[0]:
protocol_and_auth = parts[0].split("://")[1]
if ":" in protocol_and_auth:
user_pass, host_port = protocol_and_auth.split(":", 1)
return DATABASE_URL.replace(user_pass, "****")
return "****://****:****@****/****"