feat: add base FastAPI project structure with static frontend and API v1

This commit is contained in:
Lino Mallevaey
2025-08-18 00:22:50 +02:00
parent ff5026b8f7
commit 49bcb38261
14 changed files with 487 additions and 13 deletions

View File

@@ -0,0 +1,10 @@
"""
MokPyo application package.
- Central entry point for app-level imports
"""
from .main import app
from .core import settings
__version__ = settings.APP_VERSION

19
app/__main__.py Normal file
View File

@@ -0,0 +1,19 @@
"""
Entrypoint pour lancer l'application en dev.
Usage:
python3 -m app
"""
import uvicorn
from app.main import app
from app.core.config import settings
if __name__ == "__main__":
uvicorn.run(
"app.main:app",
host=settings.HOST,
port=settings.PORT,
reload=settings.ENV == "dev",
log_level=settings.LOG_LEVEL.lower(),
)

View File

@@ -0,0 +1,10 @@
"""
API v1 package for MokPyo application.
- Central router for version 1 of the API
"""
from fastapi import APIRouter
from .endpoints import router as v1_router
__version__ = "1.0.0"

View File

@@ -0,0 +1,44 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.core import settings, get_current_user, hash_password, verify_password
from app.db import get_db
from pathlib import Path
router = APIRouter()
# =========================
# TEST / PING
# =========================
@router.get("/ping", tags=["Test"])
def ping():
return {"message": "pong"}
# =========================
# EXEMPLE UTILISATEUR
# =========================
@router.get("/me", tags=["User"])
def read_current_user(current_user: dict = Depends(get_current_user)):
"""
Retourne les infos de l'utilisateur connecté
"""
return {"user": current_user}
# =========================
# EXEMPLE AUTH
# =========================
@router.post("/hash-password", tags=["Auth"])
def test_hash_password(password: str):
"""
Exemple simple pour hasher un mot de passe
"""
hashed = hash_password(password)
return {"password": password, "hashed": hashed}
@router.post("/verify-password", tags=["Auth"])
def test_verify_password(password: str, hashed: str):
"""
Vérifie qu'un mot de passe correspond à un hash
"""
valid = verify_password(password, hashed)
return {"valid": valid}

View File

@@ -0,0 +1,19 @@
"""
Core module for MokPyo application.
- Central access point for configuration and security utilities
"""
# Expose key components for easy imports
from .config import settings
from .security import (
hash_password,
verify_password,
create_access_token,
decode_access_token,
get_current_user,
set_auth_cookie,
get_cookie_token,
)
__version__ = settings.APP_VERSION

View File

@@ -0,0 +1,75 @@
from pydantic_settings import BaseSettings
from datetime import timedelta
class Settings(BaseSettings):
# =========================
# ENVIRONMENT
# =========================
ENV: str
# =========================
# DATABASE
# =========================
DATABASE_URL: str
# =========================
# FASTAPI
# =========================
DEBUG: bool = True
APP_TITLE: str = "MokPyo API"
APP_VERSION: str = "1.0.0"
APP_DESCRIPTION: str = "MokPyo"
APP_DOMAIN: str = "localhost"
STATIC_DIR: str = "app/static"
# =========================
# SERVER
# =========================
HOST: str = "0.0.0.0"
PORT: int = 8000
UVICORN_WORKERS: int = 4
# =========================
# SECURITY / AUTH
# =========================
SECRET_KEY: str
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
ALGORITHM: str = "HS256"
# =========================
# EMAIL (optionnel)
# =========================
SMTP_HOST: str = ""
SMTP_PORT: int = 587
SMTP_USER: str = ""
SMTP_PASSWORD: str = ""
EMAIL_FROM: str = ""
# =========================
# LOGGING
# =========================
LOG_LEVEL: str = "INFO"
LOG_FILE: str = "logs/app.log"
# =========================
# UPLOADS / FILES
# =========================
MAX_UPLOAD_SIZE: int = 10_485_760 # 10 MB
# =========================
# ORJSON
# =========================
ORJSON_STRICT: bool = True
@property
def access_token_expire(self) -> timedelta:
return timedelta(minutes=self.ACCESS_TOKEN_EXPIRE_MINUTES)
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
# Instance globale
settings = Settings()
print("test")

View File

@@ -0,0 +1,76 @@
from datetime import datetime, timedelta
from typing import Any, Optional
from passlib.context import CryptContext
from jose import jwt, JWTError
from fastapi import HTTPException, status, Depends, Request, Response
from fastapi.security import OAuth2PasswordBearer
from app.core.config import settings
# =========================
# PASSWORD HASHING
# =========================
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
"""Hash a password using bcrypt."""
return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""Verify a password against its hash."""
return pwd_context.verify(plain_password, hashed_password)
# =========================
# JWT TOKEN MANAGEMENT
# =========================
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/token")
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""Create a JWT access token."""
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def decode_access_token(token: str) -> dict:
"""Decode and verify a JWT access token."""
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
return payload
except JWTError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token")
# =========================
# DEPENDENCIES FASTAPI
# =========================
async def get_current_user(token: str = Depends(oauth2_scheme)) -> dict:
"""Get current user from JWT token."""
payload = decode_access_token(token)
user_id: Any = payload.get("sub")
if user_id is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token payload")
return payload
# =========================
# COOKIE MANAGEMENT (optionnel)
# =========================
def set_auth_cookie(response: Response, token: str, max_age: int = 3600):
"""Set JWT token in HttpOnly cookie."""
response.set_cookie(
key="access_token",
value=f"Bearer {token}",
httponly=True,
max_age=max_age,
samesite="lax",
secure=False, # True si HTTPS
)
def get_cookie_token(request: Request) -> Optional[str]:
"""Get JWT token from cookies."""
cookie = request.cookies.get("access_token")
if cookie and cookie.startswith("Bearer "):
return cookie[7:]
return None

View File

@@ -0,0 +1,10 @@
"""
Database package for MokPyo application.
- Expose engine and session objects for easy imports
"""
__version__ = "1.0.0"
from .engine import engine, async_engine
from .session import SessionLocal, AsyncSessionLocal, get_db, get_async_db

View File

@@ -0,0 +1,24 @@
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from sqlalchemy import create_engine
from app.core.config import settings
# =========================
# SYNCHRONOUS ENGINE
# =========================
# Pour opérations sync classiques
engine = create_engine(
settings.DATABASE_URL.replace("+aiomysql", ""), # remove async part if present
echo=settings.DEBUG,
future=True
)
# =========================
# ASYNCHRONOUS ENGINE
# =========================
# Pour opérations async avec async SQLAlchemy
async_engine: AsyncEngine = create_async_engine(
settings.DATABASE_URL if "+aiomysql" in settings.DATABASE_URL else settings.DATABASE_URL.replace("mysql+pymysql", "mysql+aiomysql"),
echo=settings.DEBUG,
future=True
)

View File

@@ -0,0 +1,45 @@
from typing import AsyncGenerator, Generator
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession
from app.db.engine import engine, async_engine
# =========================
# SYNCHRONOUS SESSION
# =========================
SessionLocal = sessionmaker(
bind=engine,
autocommit=False,
autoflush=False,
future=True
)
def get_db() -> Generator:
"""
Yield a synchronous SQLAlchemy session
Usage: Depends(get_db)
"""
db = SessionLocal()
try:
yield db
finally:
db.close()
# =========================
# ASYNCHRONOUS SESSION
# =========================
AsyncSessionLocal = sessionmaker(
bind=async_engine,
class_=AsyncSession,
expire_on_commit=False,
autocommit=False,
autoflush=False,
future=True
)
async def get_async_db() -> AsyncGenerator:
"""
Yield an asynchronous SQLAlchemy session
Usage: Depends(get_async_db)
"""
async with AsyncSessionLocal() as session:
yield session

View File

@@ -0,0 +1,49 @@
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse, JSONResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from pathlib import Path
# Import de la config
from app.core.config import settings
# Import du router v1
from app.api.v1 import v1_router
# Création de l'app FastAPI
app = FastAPI(
title=settings.APP_TITLE,
version=settings.APP_VERSION,
default_response_class=ORJSONResponse if settings.ORJSON_STRICT else JSONResponse,
debug=settings.DEBUG,
# Ajout de la documentation
openapi_url=None if settings.ENV == "prod" else "/openapi.json",
docs_url=None if settings.ENV == "prod" else "/docs",
redoc_url=None if settings.ENV == "prod" else "/redoc",
)
# =========================
# Static files
# =========================
STATIC_PATH = Path(__file__).parent / "static"
app.mount("/assets", StaticFiles(directory=STATIC_PATH / "assets"), name="assets")
# =========================
# Inclure les routes v1
# =========================
app.include_router(v1_router, prefix="/api/v1")
# =========================
# Static pages
# =========================
def add_page(endpoint: str, html_file: str):
file_path = STATIC_PATH / html_file
@app.get(endpoint, include_in_schema=False)
def page():
return FileResponse(file_path)
add_page("/", "index.html")