Add initial project configuration and setup for Reference Board Viewer application. Include EditorConfig for consistent coding styles, pre-commit hooks for linting and formatting, Docker Compose for local development with PostgreSQL and MinIO, and a Nix flake for development environment management. Establish CI/CD pipeline for automated testing and deployment.

This commit is contained in:
Danilo Reyes
2025-11-01 22:28:46 -06:00
parent 58f463867e
commit 1bc657e0fd
33 changed files with 1756 additions and 38 deletions

View File

@@ -0,0 +1,2 @@
"""Core application modules."""

View File

@@ -0,0 +1,93 @@
"""Application configuration."""
from functools import lru_cache
from typing import Any
from pydantic import PostgresDsn, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""Application settings."""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore",
)
# Application
APP_NAME: str = "Reference Board Viewer"
APP_VERSION: str = "1.0.0"
DEBUG: bool = False
API_V1_PREFIX: str = "/api/v1"
# Database
DATABASE_URL: PostgresDsn
DATABASE_POOL_SIZE: int = 20
DATABASE_MAX_OVERFLOW: int = 0
# JWT Authentication
SECRET_KEY: str
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# MinIO Storage
MINIO_ENDPOINT: str
MINIO_ACCESS_KEY: str
MINIO_SECRET_KEY: str
MINIO_BUCKET: str = "webref"
MINIO_SECURE: bool = False
# CORS
CORS_ORIGINS: list[str] = ["http://localhost:5173", "http://localhost:3000"]
@field_validator("CORS_ORIGINS", mode="before")
@classmethod
def parse_cors_origins(cls, v: Any) -> list[str]:
"""Parse CORS origins from string or list."""
if isinstance(v, str):
return [origin.strip() for origin in v.split(",")]
return v
# File Upload
MAX_FILE_SIZE: int = 52428800 # 50MB
MAX_BATCH_SIZE: int = 524288000 # 500MB
ALLOWED_MIME_TYPES: list[str] = [
"image/jpeg",
"image/png",
"image/gif",
"image/webp",
"image/svg+xml",
]
@field_validator("ALLOWED_MIME_TYPES", mode="before")
@classmethod
def parse_mime_types(cls, v: Any) -> list[str]:
"""Parse MIME types from string or list."""
if isinstance(v, str):
return [mime.strip() for mime in v.split(",")]
return v
# Performance
REQUEST_TIMEOUT: int = 30
MAX_CONCURRENT_UPLOADS: int = 10
# Security
BCRYPT_ROUNDS: int = 12
PASSWORD_MIN_LENGTH: int = 8
# Logging
LOG_LEVEL: str = "INFO"
@lru_cache
def get_settings() -> Settings:
"""Get cached application settings."""
return Settings()
# Export settings instance
settings = get_settings()

12
backend/app/core/deps.py Normal file
View File

@@ -0,0 +1,12 @@
"""Dependency injection utilities."""
from typing import Annotated, Generator
from fastapi import Depends
from sqlalchemy.orm import Session
from app.database.session import get_db
# Database session dependency
DatabaseSession = Annotated[Session, Depends(get_db)]

View File

@@ -0,0 +1,68 @@
"""Custom exception classes."""
from typing import Any
class WebRefException(Exception):
"""Base exception for all custom exceptions."""
def __init__(self, message: str, status_code: int = 500, details: dict[str, Any] | None = None):
self.message = message
self.status_code = status_code
self.details = details or {}
super().__init__(self.message)
class ValidationError(WebRefException):
"""Validation error."""
def __init__(self, message: str, details: dict[str, Any] | None = None):
super().__init__(message, status_code=422, details=details)
class AuthenticationError(WebRefException):
"""Authentication error."""
def __init__(self, message: str = "Authentication failed"):
super().__init__(message, status_code=401)
class AuthorizationError(WebRefException):
"""Authorization error."""
def __init__(self, message: str = "Insufficient permissions"):
super().__init__(message, status_code=403)
class NotFoundError(WebRefException):
"""Resource not found error."""
def __init__(self, resource: str, resource_id: str | None = None):
message = f"{resource} not found"
if resource_id:
message = f"{resource} with id {resource_id} not found"
super().__init__(message, status_code=404)
class ConflictError(WebRefException):
"""Resource conflict error."""
def __init__(self, message: str):
super().__init__(message, status_code=409)
class FileTooLargeError(WebRefException):
"""File size exceeds limit."""
def __init__(self, max_size: int):
message = f"File size exceeds maximum allowed size of {max_size} bytes"
super().__init__(message, status_code=413)
class UnsupportedFileTypeError(WebRefException):
"""Unsupported file type."""
def __init__(self, file_type: str, allowed_types: list[str]):
message = f"File type '{file_type}' not supported. Allowed types: {', '.join(allowed_types)}"
super().__init__(message, status_code=415)

View File

@@ -0,0 +1,34 @@
"""Logging configuration."""
import logging
import sys
from app.core.config import settings
def setup_logging() -> None:
"""Configure application logging."""
# Get log level from settings
log_level = getattr(logging, settings.LOG_LEVEL.upper(), logging.INFO)
# Configure root logger
logging.basicConfig(
level=log_level,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
handlers=[
logging.StreamHandler(sys.stdout)
],
)
# Set library log levels
logging.getLogger("uvicorn").setLevel(logging.INFO)
logging.getLogger("uvicorn.access").setLevel(logging.INFO)
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
logging.getLogger("boto3").setLevel(logging.WARNING)
logging.getLogger("botocore").setLevel(logging.WARNING)
logger = logging.getLogger(__name__)
logger.info(f"Logging configured with level: {settings.LOG_LEVEL}")

View File

@@ -0,0 +1,29 @@
"""CORS and other middleware configuration."""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from app.core.config import settings
def setup_middleware(app: FastAPI) -> None:
"""Configure application middleware."""
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.CORS_ORIGINS,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Security headers (optional, add more as needed)
# Note: TrustedHostMiddleware not added by default in dev
# Uncomment for production:
# app.add_middleware(
# TrustedHostMiddleware,
# allowed_hosts=["yourdomain.com", "*.yourdomain.com"]
# )

View File

@@ -0,0 +1,64 @@
"""Base Pydantic schemas."""
from datetime import datetime
from typing import Any
from uuid import UUID
from pydantic import BaseModel, ConfigDict, Field
class BaseSchema(BaseModel):
"""Base schema with common configuration."""
model_config = ConfigDict(
from_attributes=True,
populate_by_name=True,
json_schema_extra={
"example": {}
}
)
class TimestampSchema(BaseSchema):
"""Schema with timestamp fields."""
created_at: datetime = Field(..., description="Creation timestamp")
updated_at: datetime | None = Field(None, description="Last update timestamp")
class IDSchema(BaseSchema):
"""Schema with ID field."""
id: UUID = Field(..., description="Unique identifier")
class ResponseSchema(BaseSchema):
"""Generic response schema."""
message: str = Field(..., description="Response message")
data: dict[str, Any] | None = Field(None, description="Response data")
class ErrorSchema(BaseSchema):
"""Error response schema."""
error: str = Field(..., description="Error message")
details: dict[str, Any] | None = Field(None, description="Error details")
status_code: int = Field(..., description="HTTP status code")
class PaginationSchema(BaseSchema):
"""Pagination metadata schema."""
total: int = Field(..., description="Total number of items")
page: int = Field(..., description="Current page number")
page_size: int = Field(..., description="Items per page")
total_pages: int = Field(..., description="Total number of pages")
class PaginatedResponse(BaseSchema):
"""Paginated response schema."""
items: list[Any] = Field(..., description="List of items")
pagination: PaginationSchema = Field(..., description="Pagination metadata")

119
backend/app/core/storage.py Normal file
View File

@@ -0,0 +1,119 @@
"""MinIO storage client utilities."""
import logging
from io import BytesIO
from typing import BinaryIO
import boto3
from botocore.client import Config
from botocore.exceptions import ClientError
from app.core.config import settings
logger = logging.getLogger(__name__)
class StorageClient:
"""MinIO storage client wrapper."""
def __init__(self):
"""Initialize MinIO client."""
self.client = boto3.client(
"s3",
endpoint_url=f"{'https' if settings.MINIO_SECURE else 'http'}://{settings.MINIO_ENDPOINT}",
aws_access_key_id=settings.MINIO_ACCESS_KEY,
aws_secret_access_key=settings.MINIO_SECRET_KEY,
config=Config(signature_version="s3v4"),
)
self.bucket = settings.MINIO_BUCKET
self._ensure_bucket_exists()
def _ensure_bucket_exists(self) -> None:
"""Create bucket if it doesn't exist."""
try:
self.client.head_bucket(Bucket=self.bucket)
except ClientError:
logger.info(f"Creating bucket: {self.bucket}")
self.client.create_bucket(Bucket=self.bucket)
def upload_file(self, file_data: BinaryIO, object_name: str, content_type: str) -> str:
"""Upload file to MinIO.
Args:
file_data: File data to upload
object_name: S3 object name (path)
content_type: MIME type of the file
Returns:
str: Object URL
Raises:
Exception: If upload fails
"""
try:
self.client.upload_fileobj(
file_data,
self.bucket,
object_name,
ExtraArgs={"ContentType": content_type},
)
return f"{settings.MINIO_ENDPOINT}/{self.bucket}/{object_name}"
except ClientError as e:
logger.error(f"Failed to upload file {object_name}: {e}")
raise
def download_file(self, object_name: str) -> BytesIO:
"""Download file from MinIO.
Args:
object_name: S3 object name (path)
Returns:
BytesIO: File data
Raises:
Exception: If download fails
"""
try:
file_data = BytesIO()
self.client.download_fileobj(self.bucket, object_name, file_data)
file_data.seek(0)
return file_data
except ClientError as e:
logger.error(f"Failed to download file {object_name}: {e}")
raise
def delete_file(self, object_name: str) -> None:
"""Delete file from MinIO.
Args:
object_name: S3 object name (path)
Raises:
Exception: If deletion fails
"""
try:
self.client.delete_object(Bucket=self.bucket, Key=object_name)
except ClientError as e:
logger.error(f"Failed to delete file {object_name}: {e}")
raise
def file_exists(self, object_name: str) -> bool:
"""Check if file exists in MinIO.
Args:
object_name: S3 object name (path)
Returns:
bool: True if file exists, False otherwise
"""
try:
self.client.head_object(Bucket=self.bucket, Key=object_name)
return True
except ClientError:
return False
# Global storage client instance
storage_client = StorageClient()