Backend REST API được xây dựng với Python FastAPI và PostgreSQL 15+, sử dụng Layered Architecture Pattern (Kiến trúc phân lớp) chuẩn dự án thực tế.
- ✅ Layered Architecture: Kiến trúc phân lớp chuẩn (API → Service → Repository → Model)
- ✅ Chuẩn hóa Request/Response: Các schema chung cho request và response
- ✅ Exception Handling: Xử lý lỗi tập trung và chuẩn hóa
- ✅ Paging, Filter, Sort: Hỗ trợ phân trang, lọc và sắp xếp dữ liệu
- ✅ Logging: Ghi log lỗi và thay đổi database
- ✅ OpenAPI/Swagger: Tài liệu API tự động
- ✅ Testing: Cấu trúc test với pytest
- ✅ Dependency Injection: Hỗ trợ dependency injection cho services và repositories
- ✅ Base Classes: BaseRepository và BaseService với CRUD operations sẵn có
Dự án sử dụng Layered Architecture Pattern với các layer sau:
┌─────────────────────────────────────────┐
│ API/Controller Layer │ ← app/api/v1/endpoints/
│ (HTTP Request/Response Handling) │
└─────────────────┬───────────────────────┘
│
▼
┌─────────────────────────────────────────┐
│ Service Layer │ ← app/services/
│ (Business Logic) │
└─────────────────┬───────────────────────┘
│
▼
┌─────────────────────────────────────────┐
│ Repository Layer │ ← app/repositories/
│ (Data Access Logic) │
└─────────────────┬───────────────────────┘
│
▼
┌─────────────────────────────────────────┐
│ Model/Entity Layer │ ← app/db/models/
│ (Database Models) │
└─────────────────────────────────────────┘
┌─────────────────────────────────────────┐
│ Schema/DTO Layer │ ← app/schemas/
│ (Request/Response Models) │
└─────────────────────────────────────────┘
admin-api-service/
├── app/
│ ├── api/ # API/Controller Layer
│ │ └── v1/
│ │ ├── endpoints/ # API endpoints
│ │ │ ├── example.py # Example endpoint (template)
│ │ │ └── users.py # User endpoints (complete example)
│ │ └── router.py # Router chính
│ │
│ ├── services/ # Service Layer (Business Logic)
│ │ ├── base.py # BaseService với generic operations
│ │ └── user.py # UserService với business logic
│ │
│ ├── repositories/ # Repository Layer (Data Access)
│ │ ├── base.py # BaseRepository với CRUD operations
│ │ └── user.py # UserRepository với custom queries
│ │
│ ├── schemas/ # Schema/DTO Layer
│ │ ├── base.py # Base schemas (Response, Pagination)
│ │ └── user.py # User schemas (Create, Update, Response)
│ │
│ ├── db/ # Database Layer
│ │ ├── database.py # Database configuration & session
│ │ └── models/ # Model/Entity Layer
│ │ ├── base.py # BaseModel với common fields
│ │ └── user.py # User model
│ │
│ ├── core/ # Core utilities
│ │ ├── config.py # Cấu hình ứng dụng
│ │ ├── dependencies.py # Dependency injection helpers
│ │ ├── exceptions.py # Custom exceptions
│ │ ├── exception_handler.py # Exception handlers
│ │ ├── logging.py # Cấu hình logging
│ │ ├── pagination.py # Utilities phân trang
│ │ ├── filtering.py # Utilities lọc dữ liệu
│ │ ├── sorting.py # Utilities sắp xếp
│ │ └── schemas.py # (Deprecated - use app.schemas.base)
│ │
│ └── main.py # FastAPI application entry point
│
├── tests/ # Test files
├── alembic/ # Database migrations
├── requirements.txt # Python dependencies
├── ARCHITECTURE.md # Tài liệu chi tiết về kiến trúc
├── LAYER_STRUCTURE.md # Tổng quan cấu trúc layer
└── README.md
📖 Xem
ARCHITECTURE.mdđể hiểu chi tiết về từng layer và best practices.
python -m venv venv
source venv/bin/activate # Trên Windows: venv\Scripts\activatepip install -r requirements.txtTạo file .env từ .env.example và cập nhật các thông tin:
DATABASE_URL=postgresql://user:password@localhost:5432/admin_api_db
APP_NAME=Admin API Service
APP_VERSION=1.0.0
DEBUG=True
LOG_LEVEL=INFO
HOST=0.0.0.0
PORT=8000# Tạo database PostgreSQL
createdb admin_api_db
# Chạy migrations
alembic upgrade head# Development mode
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000
# Production mode
uvicorn app.main:app --host 0.0.0.0 --port 8000Sau khi chạy, truy cập:
- API Documentation: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
- OpenAPI JSON: http://localhost:8000/api/v1/openapi.json
Tất cả API responses đều tuân theo format:
{
"success": true,
"data": {...},
"message": "Success message"
}Error response:
{
"success": false,
"error": {
"code": "ERROR_CODE",
"message": "Error message",
"detail": {...}
}
}POST /api/v1/users/
Content-Type: application/json
{
"email": "user@example.com",
"full_name": "John Doe",
"password": "securepassword123",
"is_active": true
}GET /api/v1/users/?page=1&page_size=10&sort_by=created_at&sort_order=desc&is_active=trueResponse:
{
"success": true,
"data": [
{
"id": 1,
"email": "user@example.com",
"full_name": "John Doe",
"is_active": true,
"created_at": "2024-01-01T00:00:00",
"updated_at": "2024-01-01T00:00:00"
}
],
"pagination": {
"page": 1,
"page_size": 10,
"total": 100,
"total_pages": 10
}
}GET /api/v1/users/1PUT /api/v1/users/1
Content-Type: application/json
{
"full_name": "Jane Doe",
"is_active": false
}DELETE /api/v1/users/1Tất cả list endpoints hỗ trợ pagination:
GET /api/v1/users/?page=1&page_size=10Query parameters:
page: Số trang (mặc định: 1)page_size: Số items mỗi trang (mặc định: 10, tối đa: 100)
Repository layer hỗ trợ filtering với các operators:
# Simple equality
filters = {"is_active": True}
# Advanced operators
filters = {
"age": {"operator": "gt", "value": 18},
"name": {"operator": "like", "value": "John"},
"status": {"operator": "in", "value": ["active", "pending"]}
}Các operators hỗ trợ:
eq: bằng (mặc định)ne: khácgt,gte,lt,lte: so sánhlike: tìm kiếm (LIKE %value%)in: trong danh sách
GET /api/v1/users/?sort_by=email&sort_order=descQuery parameters:
sort_by: Tên field để sort (ví dụ:email,created_at,full_name)sort_order:aschoặcdesc(mặc định:asc)
Các custom exceptions:
NotFoundException(404)BadRequestException(400)UnauthorizedException(401)ForbiddenException(403)ConflictException(409)ValidationException(422)
Ví dụ sử dụng:
from app.core.exceptions import NotFoundException
raise NotFoundException("Resource not found")# Tạo migration mới
alembic revision --autogenerate -m "Description"
# Chạy migrations
alembic upgrade head
# Rollback
alembic downgrade -1# Chạy tất cả tests
pytest
# Chạy với coverage
pytest --cov=app --cov-report=html
# Chạy test cụ thể
pytest tests/test_main.pyLogging được cấu hình tự động:
- Log lỗi khi exception xảy ra
- Log thay đổi database (INSERT, UPDATE, DELETE)
- Log format JSON cho dễ parse
Theo kiến trúc Layer Architecture, mỗi module cần tạo đầy đủ các layer:
from sqlalchemy import Column, String, Numeric, Boolean
from app.db.models.base import BaseModel
class Product(BaseModel):
__tablename__ = "products"
name = Column(String(255), nullable=False, index=True)
description = Column(String(1000))
price = Column(Numeric(10, 2), nullable=False)
is_active = Column(Boolean, default=True, nullable=False)from typing import Optional
from datetime import datetime
from decimal import Decimal
from pydantic import BaseModel, Field
class ProductBase(BaseModel):
name: str = Field(..., min_length=1, max_length=255)
description: Optional[str] = None
price: Decimal = Field(..., gt=0)
is_active: bool = True
class ProductCreate(ProductBase):
pass
class ProductUpdate(BaseModel):
name: Optional[str] = None
description: Optional[str] = None
price: Optional[Decimal] = None
is_active: Optional[bool] = None
class ProductResponse(ProductBase):
id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = Truefrom typing import Optional
from sqlalchemy.orm import Session
from app.repositories.base import BaseRepository
from app.db.models.product import Product
class ProductRepository(BaseRepository[Product]):
def __init__(self, db: Session):
super().__init__(Product, db)
def get_by_name(self, name: str) -> Optional[Product]:
"""Get product by name"""
return self.get_by_field("name", name)from sqlalchemy.orm import Session
from app.services.base import BaseService
from app.repositories.product import ProductRepository
from app.db.models.product import Product
from app.schemas.product import ProductCreate, ProductUpdate
from app.core.exceptions import ConflictException
class ProductService(BaseService[Product, ProductRepository]):
def __init__(self, db: Session):
repository = ProductRepository(db)
super().__init__(repository)
def create_product(self, product_data: ProductCreate) -> Product:
"""Create product with business logic validation"""
# Business logic: Check if name already exists
if self.repository.get_by_name(product_data.name):
raise ConflictException(f"Product with name '{product_data.name}' already exists")
return self.repository.create(**product_data.model_dump())
def update_product(self, product_id: int, product_data: ProductUpdate) -> Product:
"""Update product with business logic validation"""
update_data = product_data.model_dump(exclude_unset=True)
return self.repository.update(product_id, **update_data)from fastapi import APIRouter, Depends, Query, status
from sqlalchemy.orm import Session
from app.db.database import get_db
from app.services.product import ProductService
from app.schemas.base import BaseResponse, PaginatedResponse
from app.schemas.product import ProductCreate, ProductUpdate, ProductResponse
from app.schemas.base import PaginationParams, SortParams
router = APIRouter()
def get_product_service(db: Session = Depends(get_db)) -> ProductService:
"""Dependency to get ProductService instance"""
return ProductService(db)
@router.get("/", response_model=PaginatedResponse[ProductResponse])
async def list_products(
page: int = Query(1, ge=1),
page_size: int = Query(10, ge=1, le=100),
sort_by: str = Query(None),
sort_order: str = Query("asc", pattern="^(asc|desc)$"),
service: ProductService = Depends(get_product_service)
):
"""List products with pagination and sorting"""
pagination = PaginationParams(page=page, page_size=page_size)
sort_params = SortParams(sort_by=sort_by, sort_order=sort_order)
result = service.get_paginated(
pagination=pagination,
sort_params=sort_params
)
return result
@router.post("/", response_model=BaseResponse[ProductResponse], status_code=status.HTTP_201_CREATED)
async def create_product(
product_data: ProductCreate,
service: ProductService = Depends(get_product_service)
):
"""Create a new product"""
product = service.create_product(product_data)
return BaseResponse(data=product, message="Product created successfully")
@router.get("/{product_id}", response_model=BaseResponse[ProductResponse])
async def get_product(
product_id: int,
service: ProductService = Depends(get_product_service)
):
"""Get a single product by ID"""
product = service.get_by_id_or_raise(product_id)
return BaseResponse(data=product, message="Product retrieved successfully")
@router.put("/{product_id}", response_model=BaseResponse[ProductResponse])
async def update_product(
product_id: int,
product_data: ProductUpdate,
service: ProductService = Depends(get_product_service)
):
"""Update an existing product"""
product = service.update_product(product_id, product_data)
return BaseResponse(data=product, message="Product updated successfully")
@router.delete("/{product_id}", response_model=BaseResponse)
async def delete_product(
product_id: int,
service: ProductService = Depends(get_product_service)
):
"""Delete a product"""
service.delete(product_id)
return BaseResponse(message="Product deleted successfully")from app.api.v1.endpoints import products
api_router.include_router(products.router, prefix="/products", tags=["products"])# Import model trong alembic/env.py
from app.db.models.product import Product
# Tạo migration
alembic revision --autogenerate -m "Add Product model"
alembic upgrade head💡 Tip: Xem module
User(app/api/v1/endpoints/users.py) để tham khảo implementation hoàn chỉnh.
Ví dụ: Tạo User mới
1. HTTP Request: POST /api/v1/users/
↓
2. API Layer (app/api/v1/endpoints/users.py)
- Validate request body với UserCreate schema
- Extract user_data từ request
↓
3. Service Layer (app/services/user.py)
- Business logic: Check email exists
- Business logic: Hash password
- Call repository.create()
↓
4. Repository Layer (app/repositories/user.py)
- Build SQL INSERT query
- Execute query
↓
5. Database
- Insert record
- Return created record
↓
6. Repository → Service → API
- Return User model → Transform → Serialize to UserResponse
↓
7. HTTP Response: 201 Created với UserResponse
-
Dependency Direction: Luôn tuân thủ hướng phụ thuộc
- API → Service → Repository → Model
- Không bao giờ ngược lại
-
Business Logic: Tất cả business logic phải ở Service layer
- Không đặt business logic ở Repository hoặc API layer
-
Data Access: Chỉ Repository layer truy cập database
- API layer không bao giờ truy cập database trực tiếp
-
Error Handling: Sử dụng custom exceptions từ
app.core.exceptions -
Type Hints: Luôn sử dụng type hints để tăng code clarity
ARCHITECTURE.md: Tài liệu chi tiết về kiến trúc và từng layerLAYER_STRUCTURE.md: Tổng quan cấu trúc và cách sử dụng- Module
User: Ví dụ implementation hoàn chỉnh
- Python 3.10+
- PostgreSQL 15+
- pip
- FastAPI: Web framework
- SQLAlchemy: ORM
- Pydantic: Data validation
- Alembic: Database migrations
- Passlib: Password hashing
- Pytest: Testing framework
MIT