explorigin/loapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: explorigin:35314f4f8b7e8efd6e4542e4799550cbd8fbb8c2
Branches Tags
explorigin:master
explorigin:payoff-test
explorigin:rate-limiting
..
compare: explorigin:a5f96d071092e97159c1d7b6772e87b81b8d2a65
Branches Tags
explorigin:payoff-test
explorigin:master
explorigin:rate-limiting

No commits in common. "35314f4f8b7e8efd6e4542e4799550cbd8fbb8c2" and "a5f96d071092e97159c1d7b6772e87b81b8d2a65" have entirely different histories.
35314f4f8b ... a5f96d0710

17 changed files with 4 additions and 2334 deletions
Show Stats Expand all files Collapse all files

9
.gitignore vendored
View File

@ -1,10 +1 @@
.aider* .aider*
.venv
__pycache__
*.pyc
.mypy_cache
.env

7
.vscode/settings.json vendored
View File

@ -1,7 +0,0 @@
{
"python.testing.pytestArgs": [
"tests"
],
"python.testing.unittestEnabled": false,
"python.testing.pytestEnabled": true
}

55
.vscode/tasks.json vendored
View File

@ -1,55 +0,0 @@
{
// See https://go.microsoft.com/fwlink/?LinkId=733558
// for the documentation about the tasks.json format
"version": "2.0.0",
"tasks": [
{
"label": "Run Mypy",
"type": "shell",
"command": "uv run mypy . --no-incremental",
"problemMatcher": [
"$python"
]
},
{
"label": "Isort Check",
"type": "shell",
"command": "uv run isort . --check --diff",
"problemMatcher": [
"$python"
]
},
{
"label": "Isort Fix",
"type": "shell",
"command": "uv run isort .",
"problemMatcher": [
"$python"
]
},
{
"label": "AutoPEP8 Check",
"type": "shell",
"command": "uv run autopep8 app/**/*.py --diff",
"problemMatcher": [
"$python"
]
},
{
"label": "AutoPEP8 Fix",
"type": "shell",
"command": "uv run autopep8 app/**/*.py --in-place",
"problemMatcher": [
"$python"
]
},
{
"label": "Autoflake Fix",
"type": "shell",
"command": "uv run autoflake -i app/**/*.py",
"problemMatcher": [
"$python"
]
}
]
}

26
README.md
View File

@ -1,26 +0,0 @@
# Loan Operations API (loapi)
A FastAPI-based loan operations management system.
## Features
- RESTful API for loan operations
- Automatic API documentation
- Request/response validation
- Middleware for logging and error handling
## Development
```bash
# Install dependencies
uv sync
# Run the development server
uv run python main.py
```
## API Documentation
Once running, visit:
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc

0
app/__init__.py
View File

34
app/config.py
View File

@ -1,34 +0,0 @@
from pydantic import Field
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
env: str = Field(default='DEV', alias="ENV")
# 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")
# Database settings
oracle_user: str = Field(alias="ORACLE_USER")
oracle_password: str = Field(alias="ORACLE_PASSWORD")
oracle_dsn: str = Field(alias="ORACLE_DSN")
# Sentry settings
sentry_dsn: str | None = Field(default=None, alias="SENTRY_DSN")
# 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")
model_config = {
"env_file": ".env",
"case_sensitive": False
}
settings = Settings() # type:ignore[call-arg]

50
app/database.py
View File

@ -1,50 +0,0 @@
"""Database configuration and connection management."""
from typing import Generator
from sqlalchemy import MetaData, create_engine
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from app.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() -> Generator[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

42
app/logging.py
View File

@ -1,42 +0,0 @@
"""Logging configuration with structured logging."""
import logging
import sys
import structlog
from app.config import settings
def configure_logging() -> None:
"""Configure structured logging."""
structlog.configure(
processors=[
structlog.stdlib.filter_by_level,
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.TimeStamper(fmt="iso"),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
structlog.processors.JSONRenderer(
) if not settings.debug else structlog.dev.ConsoleRenderer(),
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
# Configure standard logging
logging.basicConfig(
format="%(message)s",
stream=sys.stdout,
level=getattr(logging, settings.log_level.upper()),
)
def get_logger(name: str) -> structlog.BoundLogger:
"""Get a configured logger instance."""
return structlog.get_logger(name) # type:ignore[no-any-return]

36
app/middleware.py
View File

@ -1,36 +0,0 @@
from typing import Callable, Any
from fastapi import Request, Response
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.middleware import SlowAPIMiddleware
from slowapi.util import get_remote_address
from app.config import settings
from app.logging import get_logger
logger = get_logger(__name__)
# Rate limiter
limiter = Limiter(key_func=get_remote_address)
async def logging_middleware(request: Request, call_next: Callable[[Request], Any]) -> Any:
"""Log all requests and responses."""
logger.debug(
"Request received",
method=request.method,
url=str(request.url),
client_ip=get_remote_address(request)
)
response = await call_next(request)
logger.debug(
"Response sent",
status_code=response.status_code,
method=request.method,
url=str(request.url)
)
return response

0
app/resources/__init__.py
View File

30
app/resources/health.py
View File

@ -1,30 +0,0 @@
"""Health check endpoints."""
from fastapi import APIRouter
from app.database import DatabaseService
from app.logging import get_logger
router = APIRouter()
logger = get_logger(__name__)
@router.get("/health-check")
async def health_check() -> dict[str, str]:
"""Comprehensive health check endpoint."""
# Check database connectivity
try:
db_healthy = DatabaseService.health_check()
except Exception as e:
logger.error("Database health check failed", exc_info=e)
db_healthy = False
status = "healthy" if db_healthy else "unhealthy"
result = {
"status": status,
"service": "loapi",
"database": "connected" if db_healthy else "disconnected"
}
return result

73
app/resources/payoff.py
View File

@ -1,73 +0,0 @@
"""Pay-off endpoints."""
from datetime import date
from decimal import Decimal
from enum import Enum
from typing import Optional
from fastapi import APIRouter
from pydantic import BaseModel, Field, validator
from app.logging import get_logger
router = APIRouter()
logger = get_logger(__name__)
class PayoffStatus(str, Enum):
"""Enum for payoff status values."""
PROCESSED = "processed"
PENDING = "pending"
FAILED = "failed"
CANCELLED = "cancelled"
class PayoffRequest(BaseModel):
"""Request model for loan payoff."""
loan_id: str = Field(..., description="Loan identifier", min_length=1, max_length=50)
payoff_amount: Decimal = Field(..., description="Payoff amount", gt=0, decimal_places=2)
payoff_date: date = Field(..., description="Payoff date")
borrower_name: Optional[str] = Field(None, description="Borrower name", max_length=100)
payment_method: str = Field(..., description="Payment method", regex="^(check|wire|ach|cash)$")
notes: Optional[str] = Field(None, description="Additional notes", max_length=500)
@validator('payoff_date')
def validate_payoff_date(cls, v):
"""Validate payoff date is not in the future."""
from datetime import date
if v > date.today():
raise ValueError('Payoff date cannot be in the future')
return v
class PayoffResponse(BaseModel):
"""Response model for loan payoff."""
status: PayoffStatus
loan_id: str
payoff_amount: Decimal
payoff_date: date
transaction_id: str
@router.post("/payoff", response_model=PayoffResponse)
async def create_payoff(payoff_request: PayoffRequest) -> PayoffResponse:
"""Create a loan payoff record."""
logger.info(f"Processing payoff for loan {payoff_request.loan_id}")
# TODO: Add database logic to process payoff
# For now, return a mock response
import uuid
transaction_id = str(uuid.uuid4())
response = PayoffResponse(
status=PayoffStatus.PROCESSED,
loan_id=payoff_request.loan_id,
payoff_amount=payoff_request.payoff_amount,
payoff_date=payoff_request.payoff_date,
transaction_id=transaction_id
)
logger.info(f"Payoff processed successfully for loan {payoff_request.loan_id}, transaction {transaction_id}")
return response

16
example.env
View File

@ -1,16 +0,0 @@
# Server settings
HOST=0.0.0.0
PORT=8000
DEBUG=true
LOG_LEVEL=INFO
# Database settings
ORACLE_USER=your_oracle_user
ORACLE_PASSWORD=your_oracle_password
ORACLE_DSN=your_oracle_dsn
# Monitoring
SENTRY_DSN=
DD_SERVICE=
DD_ENV=
DD_VERSION=

65
main.py
View File

@ -1,67 +1,6 @@
""" def main():
Main entry point for the Loan Operations API. print("Hello from loapi!")
"""
import sentry_sdk
from ddtrace import patch_all
from fastapi import APIRouter, FastAPI
from sentry_sdk.integrations.fastapi import FastApiIntegration
from app.config import settings
from app.logging import configure_logging, get_logger
from app.middleware import logging_middleware
from app.resources import health, payoff
def create_app() -> FastAPI:
"""Create and configure the FastAPI application."""
# Configure monitoring
if settings.sentry_dsn:
sentry_sdk.init(
dsn=settings.sentry_dsn,
integrations=[FastApiIntegration()],
traces_sample_rate=1.0,
environment=settings.dd_env,
)
if settings.dd_service:
# Configure Datadog tracing
patch_all()
# Configure logging
configure_logging()
logger = get_logger(__name__)
# Create FastAPI app
app = FastAPI(
title="Loan Operations API",
description="SBA Loan Operations API",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc",
)
# Include all endpoint routers
app.include_router(health.router, tags=["health"])
app.include_router(payoff.router, tags=["payoff"])
return app
def main() -> None:
"""Run the application."""
import uvicorn
uvicorn.run(
app,
host=settings.host,
port=settings.port,
log_level=settings.log_level.lower(),
reload=settings.debug,
)
app = create_app()
if __name__ == "__main__": if __name__ == "__main__":
main() main()

62
pyproject.toml
View File

@ -3,63 +3,5 @@ name = "loapi"
version = "1.0.0" version = "1.0.0"
description = "Loan Operations API" description = "Loan Operations API"
readme = "README.md" readme = "README.md"
requires-python = ">=3.11" requires-python = ">=3.14"
dependencies = [ dependencies = []
"fastapi>=0.104.0",
"uvicorn[standard]>=0.24.0",
"sqlalchemy>=2.0.0",
"oracledb>=1.4.0",
"pydantic>=2.5.0",
"pydantic-settings>=2.1.0",
"slowapi>=0.1.9",
"sentry-sdk[fastapi]>=1.38.0",
"ddtrace>=2.5.0",
"structlog>=23.2.0",
"alembic>=1.13.0",
]
[project.optional-dependencies]
dev = [
"pytest>=7.4.0",
"pytest-asyncio>=0.21.0",
"pytest-cov>=4.1.0",
"mypy>=1.7.0",
"pylint>=3.0.0",
"autopep8>=2.0.0",
"isort>=5.12.0",
"types-requests>=2.31.0",
"httpx>=0.25.0",
]
[tool.mypy]
python_version = "3.14"
strict = true
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
exclude = ["tests"]
[tool.pylint.messages_control]
max-line-length = 88
disable = ["C0114", "C0116"]
[tool.isort]
profile = "black"
multi_line_output = 3
line_length = 88
[tool.pytest.ini_options]
pythonpath = ["."]
testpaths = ["tests"]
[dependency-groups]
dev = [
"autoflake>=2.3.1",
"autopep8>=2.3.2",
"httpx>=0.28.1",
"isort>=7.0.0",
"mypy>=1.19.1",
"pylint-pydantic>=0.4.1",
"pylint[pydantic]>=4.0.4",
"pytest[coverage]>=9.0.2",
]

61
tests/test_health.py
View File

@ -1,61 +0,0 @@
"""Tests for health check endpoints."""
import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch
from main import create_app
class TestHealthCheck:
"""Test cases for health check endpoint."""
@pytest.fixture
def client(self) -> TestClient:
"""Create test client."""
app = create_app()
return TestClient(app)
@patch('app.database.DatabaseService.health_check')
def test_health_check_database_connected(self, mock_db_health, client) -> None:
"""Test health check when database is connected."""
# Mock database as healthy
mock_db_health.return_value = True
response = client.get("/health-check")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert data["service"] == "loapi"
assert data["database"] == "connected"
mock_db_health.assert_called_once()
@patch('app.database.DatabaseService.health_check')
def test_health_check_database_disconnected(self, mock_db_health, client):
"""Test health check when database is disconnected."""
# Mock database as unhealthy
mock_db_health.return_value = False
response = client.get("/health-check")
assert response.status_code == 200
data = response.json()
assert data["status"] == "unhealthy"
assert data["service"] == "loapi"
assert data["database"] == "disconnected"
mock_db_health.assert_called_once()
@patch('app.database.DatabaseService.health_check')
def test_health_check_database_exception(self, mock_db_health, client):
"""Test health check when database check raises exception."""
# Mock database health check to raise exception
mock_db_health.side_effect = Exception("Database connection failed")
response = client.get("/health-check")
assert response.status_code == 200
data = response.json()
assert data["status"] == "unhealthy"
assert data["service"] == "loapi"
assert data["database"] == "disconnected"
mock_db_health.assert_called_once()

1772
uv.lock generated
View File

File diff suppressed because it is too large Load Diff