Cleanup boilerplate
This commit is contained in:
parent
5fc035b5df
commit
cd2f13caf2
0
app/core/__init__.py
Normal file
0
app/core/__init__.py
Normal file
@ -1,39 +1,32 @@
|
||||
"""
|
||||
Core configuration settings for the Loan Operations API.
|
||||
"""
|
||||
from typing import List
|
||||
from pydantic import Field
|
||||
from pydantic_settings import BaseSettings
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
"""Application settings."""
|
||||
"""Application settings following 12-factor app principles."""
|
||||
|
||||
# API Configuration
|
||||
api_title: str = "Loan Operations API"
|
||||
api_description: str = "A comprehensive API for managing loan operations"
|
||||
api_version: str = "1.0.0"
|
||||
# Server settings
|
||||
host: str = Field(default="0.0.0.0", alias="HOST")
|
||||
port: int = Field(default=8000, alias="PORT")
|
||||
debug: bool = Field(default=False, alias="DEBUG")
|
||||
log_level: str = Field(default="INFO", alias="LOG_LEVEL")
|
||||
|
||||
# Server Configuration
|
||||
host: str = Field(default="0.0.0.0", description="Server host")
|
||||
port: int = Field(default=8000, description="Server port")
|
||||
debug: bool = Field(default=False, description="Debug mode")
|
||||
# Database settings
|
||||
oracle_user: str = Field(alias="ORACLE_USER")
|
||||
oracle_password: str = Field(alias="ORACLE_PASSWORD")
|
||||
oracle_dsn: str = Field(alias="ORACLE_DSN")
|
||||
|
||||
# API Configuration
|
||||
api_v1_prefix: str = "/api/v1"
|
||||
# Sentry settings
|
||||
sentry_dsn: str | None = Field(default=None, alias="SENTRY_DSN")
|
||||
|
||||
# CORS Configuration
|
||||
allowed_origins: list[str] = Field(
|
||||
default=["*"],
|
||||
description="Allowed CORS origins"
|
||||
)
|
||||
|
||||
# Logging Configuration
|
||||
log_level: str = Field(default="INFO", description="Logging level")
|
||||
# Datadog settings
|
||||
dd_service: str = Field(default="loapi", alias="DD_SERVICE")
|
||||
dd_env: str = Field(default="development", alias="DD_ENV")
|
||||
dd_version: str = Field(default="1.0.0", alias="DD_VERSION")
|
||||
|
||||
class Config:
|
||||
env_file = ".env"
|
||||
case_sensitive = False
|
||||
|
||||
|
||||
# Global settings instance
|
||||
settings = Settings()
|
||||
settings = Settings() # type:ignore[call-arg]
|
||||
|
||||
48
app/core/database.py
Normal file
48
app/core/database.py
Normal file
@ -0,0 +1,48 @@
|
||||
"""Database configuration and connection management."""
|
||||
from typing import AsyncGenerator
|
||||
from sqlalchemy import create_engine, MetaData
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
|
||||
from sqlalchemy.orm import DeclarativeBase, sessionmaker, Session
|
||||
from app.core.config import settings
|
||||
|
||||
|
||||
class Base(DeclarativeBase):
|
||||
"""Base class for SQLAlchemy models."""
|
||||
metadata = MetaData()
|
||||
|
||||
|
||||
# Database URL construction
|
||||
DATABASE_URL = f"oracle+oracledb://{settings.oracle_user}:{settings.oracle_password}@{settings.oracle_dsn}"
|
||||
|
||||
# Synchronous engine for Oracle
|
||||
engine = create_engine(
|
||||
DATABASE_URL,
|
||||
echo=settings.debug,
|
||||
pool_pre_ping=True,
|
||||
pool_recycle=3600,
|
||||
)
|
||||
|
||||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
|
||||
def get_db() -> AsyncGenerator[Session, None]:
|
||||
"""Dependency to get database session."""
|
||||
db = SessionLocal()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
class DatabaseService:
|
||||
"""Database service for health checks and utilities."""
|
||||
|
||||
@staticmethod
|
||||
def health_check() -> bool:
|
||||
"""Check database connectivity."""
|
||||
try:
|
||||
with SessionLocal() as db:
|
||||
db.execute("SELECT 1 FROM DUAL") # type:ignore[call-overload]
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
0
app/resources/__init__.py
Normal file
0
app/resources/__init__.py
Normal file
23
app/resources/health.py
Normal file
23
app/resources/health.py
Normal file
@ -0,0 +1,23 @@
|
||||
"""Health check endpoints."""
|
||||
from fastapi import APIRouter
|
||||
from app.core.database import DatabaseService
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/health-check")
|
||||
async def health_check() -> dict[str, str]:
|
||||
"""Comprehensive health check endpoint."""
|
||||
|
||||
# Check database connectivity
|
||||
db_healthy = DatabaseService.health_check()
|
||||
|
||||
status = "healthy" if db_healthy else "unhealthy"
|
||||
|
||||
result = {
|
||||
"status": status,
|
||||
"service": "loapi",
|
||||
"database": "connected" if db_healthy else "disconnected"
|
||||
}
|
||||
|
||||
return result
|
||||
27
main.py
27
main.py
@ -6,40 +6,28 @@ from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.core.config import settings
|
||||
from app.resources import health
|
||||
|
||||
|
||||
def create_app() -> FastAPI:
|
||||
"""Create and configure the FastAPI application."""
|
||||
|
||||
app = FastAPI(
|
||||
title=settings.api_title,
|
||||
description=settings.api_description,
|
||||
version=settings.api_version,
|
||||
title="Loan Operations API",
|
||||
description="SBA Loan Operations API",
|
||||
version="1.0.0",
|
||||
docs_url="/docs",
|
||||
redoc_url="/redoc",
|
||||
)
|
||||
|
||||
# Add CORS middleware
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=settings.allowed_origins,
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
# Health check endpoint
|
||||
@app.get("/health-check")
|
||||
async def health_check():
|
||||
"""Health check endpoint."""
|
||||
return {"status": "healthy", "service": "loapi"}
|
||||
# Include all endpoint routers
|
||||
app.include_router(health.router, tags=["health"])
|
||||
|
||||
return app
|
||||
|
||||
|
||||
def main():
|
||||
def main() -> None:
|
||||
"""Run the application."""
|
||||
app = create_app()
|
||||
|
||||
uvicorn.run(
|
||||
app,
|
||||
@ -49,6 +37,7 @@ def main():
|
||||
reload=settings.debug,
|
||||
)
|
||||
|
||||
app = create_app()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user