Initial commit

OOP_FLOW.md

+# FastAPI OOP Layered Architecture Flow
+
+This document explains the request/response flow in our Object-Oriented FastAPI application. We use a **Controller-Service-Repository** pattern to ensure clean separation of concerns.
+
+## The Flow at a Glance
+
+```mermaid
+sequenceDiagram
+    participant Client
+    participant Controller as API Layer (Controller)
+    participant Service as Service Layer
+    participant Repository as Repository Layer
+    participant Database as SQLModel/Database
+
+    Client->>Controller: HTTP Request (e.g., POST /items)
+    Note over Controller: Validates Input (Pydantic/SQLModel)
+    Controller->>Service: Call Business Logic (service.create)
+    Note over Service: Applies Business Rules
+    Service->>Repository: Clean Data for DB (repository.create)
+    Note over Repository: Translates to SQL
+    Repository->>Database: Execute SQL Query
+    Database-->>Repository: Return Raw Data/Row
+    Repository-->>Service: Return ORM Object
+    Service-->>Controller: Return Model
+    Controller-->>Client: HTTP Response (JSON)
+```
+
+## Detailed Layer Breakdown
+
+### 1. API Layer (Controller)
+**Location**: `app/api/v1/endpoints/item.py`
+
+*   **Responsibility**: Handles HTTP requests, parses query parameters/bodies, and validates data types using Pydantic/SQLModel schemas (`ItemCreate`, `ItemUpdate`).
+*   **Action**: It purely orchestrates. It does *not* contain business logic or database queries. It injects the `Service` using FastAPI's dependency injection (`Depends`).
+*   **Example**:
+    ```python
+    @router.post("/")
+    def create_item(item_in: ItemCreate, service: ItemService = Depends(get_item_service)):
+        # 1. Validation happens automatically via item_in
+        # 2. Convert input to Model (optional, handled here or service)
+        item = Item.model_validate(item_in)
+        # 3. Delegate to Service
+        return service.create(item)
+    ```
+
+### 2. Service Layer
+**Location**: `app/services/item.py` (inherits from `app/services/base.py`)
+
+*   **Responsibility**: Contains the **business logic**. If you need to send an email after creating an item, calculation checks, or complex validations that involve multiple data sources, it happens here.
+*   **Action**: It receives domain models or data from the Controller and calls the Repository. It creates a transaction boundary (though in this simple app, transaction commit happens closer to the repo for simplicity, but logically it belongs here).
+*   **Example**:
+    ```python
+    class ItemService(BaseService[Item, ItemRepository]):
+        # Inherits generic CRUD methods from BaseService
+        # You can add specific logic here:
+        # def create(self, obj_in):
+        #     if obj_in.title == "Forbidden": raise Error...
+        #     return super().create(obj_in)
+        pass
+    ```
+
+### 3. Repository Layer
+**Location**: `app/repositories/item.py` (inherits from `app/repositories/base.py`)
+
+*   **Responsibility**: **Data Access**. It knows *how* to talk to the database (SQLAlchemy/SQLModel). It shouldn't know about HTTP requests or complex business rules.
+*   **Action**: Performs `select`, `insert`, `update`, `delete` operations directly on the database session.
+*   **Example**:
+    ```python
+    class BaseRepository(Generic[ModelType]):
+        def create(self, obj_in: ModelType) -> ModelType:
+            self.session.add(obj_in)
+            self.session.commit() # Persist to DB
+            self.session.refresh(obj_in) # Reload with ID and defaults
+            return obj_in
+    ```
+
+### 4. Database (Model)
+**Location**: `app/models/item.py`
+
+*   **Responsibility**: Defines the data structure (Schema) matching the database table.
+*   **Action**: SQLModel translates these classes into SQL `CREATE TABLE` commands and rows.
+
+## Why this structure?
+
+1.  **Decoupling**: You can change the database (e.g., to MongoDB) by only changing the **Repository**, without touching the Controller or Service.
+2.  **Testing**: You can easily unit test the **Service** by mocking the **Repository**. You don't need a real database running to test business logic.
+3.  **Visual Flow**: `Request` -> `Controller` -> `Service` -> `Repository` -> `DB`.
+
+## OOP Principles in Action
+
+### 1. Inheritance (Code Reuse)
+We use inheritance to avoid repeating code for common operations.
+- **Base Class**: [BaseRepository](file:///d:/Work/personal/fast-api/oop/app/repositories/base.py) defines the logic for `create`, `get_all`, `update`, and `delete`.
+- **Child Class**: [ItemRepository](file:///d:/Work/personal/fast-api/oop/app/repositories/item.py) inherits from `BaseRepository`. It gets all that functionality automatically.
+- **Benefit**: If you add a `User` model, you just create `class UserRepository(BaseRepository[User])` and you are done.
+
+### 2. Generics (Type Safety & Polymorphism)
+We use Python's `Generic` type system to make our base classes adaptable.
+- The `BaseRepository` interacts with a generic `ModelType`.
+- When we define `ItemRepository(BaseRepository[Item])`, we adhere to the Liskov Substitution Principle—`ItemRepository` is a valid specialized version of the generic repository.
+
+### 3. Encapsulation (Separation of Concerns)
+Each class hides its internal complexity:
+- **Service Layer**: The `ItemService` encapsulates *business rules*. The controller calls `service.create(item)` without knowing if that involves checking a blacklist, sending an email, or just saving to DB.
+- **Repository Layer**: Encapsulates *SQL logic*. The service doesn't know we are using SQLModel. We could swap this for a strictly SQLAlchemy or even a MongoDB implementation (with some adjustments) without changing the Service code interactively.

app/api/deps.py

+from fastapi import Depends
+from sqlmodel import Session
+from app.core.database import get_session
+from app.repositories.item import ItemRepository
+from app.services.item import ItemService
+
+def get_item_repository(session: Session = Depends(get_session)) -> ItemRepository:
+    return ItemRepository(session)
+
+def get_item_service(repository: ItemRepository = Depends(get_item_repository)) -> ItemService:
+    return ItemService(repository)

app/api/v1/endpoints/item.py

+from typing import List
+from fastapi import APIRouter, Depends, HTTPException
+from app.models.item import Item
+from app.schemas.item import ItemCreate, ItemResponse, ItemUpdate
+from app.services.item import ItemService
+from app.api.deps import get_item_service
+
+router = APIRouter()
+
+@router.post("/", response_model=ItemResponse)
+def create_item(
+    item_in: ItemCreate,
+    service: ItemService = Depends(get_item_service)
+):
+    """
+    Create a new item.
+
+    Args:
+        item_in (ItemCreate): The item data to create.
+        service (ItemService): The injected service.
+
+    Returns:
+        ItemResponse: The created item.
+    """
+    item = Item.model_validate(item_in)
+    return service.create(item)
+
+@router.get("/", response_model=List[ItemResponse])
+def read_items(
+    service: ItemService = Depends(get_item_service)
+):
+    return service.get_all()
+
+@router.get("/{item_id}", response_model=ItemResponse)
+def read_item(
+    item_id: int,
+    service: ItemService = Depends(get_item_service)
+):
+    item = service.get_by_id(item_id)
+    if not item:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return item
+
+@router.put("/{item_id}", response_model=ItemResponse)
+def update_item(
+    item_id: int,
+    item_in: ItemUpdate,
+    service: ItemService = Depends(get_item_service)
+):
+    item = service.update(item_id, item_in.model_dump(exclude_unset=True))
+    if not item:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return item
+
+@router.delete("/{item_id}", response_model=ItemResponse)
+def delete_item(
+    item_id: int,
+    service: ItemService = Depends(get_item_service)
+):
+    item = service.delete(item_id)
+    if not item:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return item

app/core/database.py

+from sqlmodel import SQLModel, create_engine, Session
+from collections.abc import Generator
+
+sqlite_file_name = "database.db"
+sqlite_url = f"sqlite:///{sqlite_file_name}"
+
+connect_args = {"check_same_thread": False}
+engine = create_engine(sqlite_url, connect_args=connect_args)
+
+def create_db_and_tables():
+    SQLModel.metadata.create_all(engine)
+
+def get_session() -> Generator[Session, None, None]:
+    with Session(engine) as session:
+        yield session

app/main.py

+from fastapi import FastAPI
+from app.api.v1.endpoints import item
+from app.core.database import create_db_and_tables
+
+app = FastAPI(title="FastAPI OOP CRUD")
+
+@app.on_event("startup")
+def on_startup():
+    create_db_and_tables()
+
+app.include_router(item.router, prefix="/items", tags=["items"])
+
+@app.get("/")
+def read_root():
+    return {"message": "Welcome to FastAPI OOP CRUD"}

app/models/base.py

+from datetime import datetime
+from typing import Optional
+from sqlmodel import Field, SQLModel
+
+class BaseModel(SQLModel):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)

app/models/item.py

+from typing import Optional
+from sqlmodel import Field
+from app.models.base import BaseModel
+
+class Item(BaseModel, table=True):
+    """
+    Represents an Item in the system.
+    
+    Attributes:
+        title (str): The name of the item.
+        description (Optional[str]): A detailed description of the item.
+        is_completed (bool): Whether the item is marked as complete.
+    """
+    title: str = Field(index=True)
+    description: Optional[str] = Field(default=None)
+    is_completed: bool = Field(default=False)

app/repositories/base.py

+from typing import Generic, TypeVar, Type, List, Optional
+from sqlmodel import Session, select, SQLModel
+
+ModelType = TypeVar("ModelType", bound=SQLModel)
+
+class BaseRepository(Generic[ModelType]):
+    """
+    A generic repository class providing common database operations for a specific model.
+
+    This class abstracts the underlying database interactions (using SQLModel/SQLAlchemy)
+    providing a clean interface for the service layer.
+
+    Attributes:
+        model (Type[ModelType]): The SQLModel class this repository operates on.
+        session (Session): The database session.
+    """
+    def __init__(self, model: Type[ModelType], session: Session):
+        """
+        Initializes the repository.
+
+        Args:
+            model (Type[ModelType]): The SQLModel class.
+            session (Session): The active database session.
+        """
+        self.model = model
+        self.session = session
+
+    def get_all(self) -> List[ModelType]:
+        """
+        Retrieves all records for the model.
+
+        Returns:
+            List[ModelType]: A list of model instances.
+        """
+        statement = select(self.model)
+        results = self.session.exec(statement)
+        return results.all()
+
+    def get_by_id(self, id: int) -> Optional[ModelType]:
+        """
+        Retrieves a single record by its ID.
+
+        Args:
+            id (int): The primary key of the record.
+
+        Returns:
+            Optional[ModelType]: The model instance if found, else None.
+        """
+        return self.session.get(self.model, id)
+
+    def create(self, obj_in: ModelType) -> ModelType:
+        """
+        Creates a new record in the database.
+
+        Args:
+            obj_in (ModelType): The model instance to create.
+
+        Returns:
+            ModelType: The created model instance with updated fields (e.g. ID).
+        """
+        self.session.add(obj_in)
+        self.session.commit()
+        self.session.refresh(obj_in)
+        return obj_in
+
+    def update(self, db_obj: ModelType, obj_in_data: dict) -> ModelType:
+        """
+        Updates an existing record.
+
+        Args:
+            db_obj (ModelType): The existing database object.
+            obj_in_data (dict): A dictionary of fields to update.
+
+        Returns:
+            ModelType: The updated model instance.
+        """
+        for key, value in obj_in_data.items():
+            setattr(db_obj, key, value)
+        self.session.add(db_obj)
+        self.session.commit()
+        self.session.refresh(db_obj)
+        return db_obj
+
+    def delete(self, db_obj: ModelType) -> ModelType:
+        """
+        Deletes a record from the database.
+
+        Args:
+            db_obj (ModelType): The database object to delete.
+
+        Returns:
+            ModelType: The deleted object (useful for returning what was deleted).
+        """
+        self.session.delete(db_obj)
+        self.session.commit()
+        return db_obj

app/repositories/item.py

+from sqlmodel import Session
+from app.models.item import Item
+from app.repositories.base import BaseRepository
+
+class ItemRepository(BaseRepository[Item]):
+    def __init__(self, session: Session):
+        super().__init__(Item, session)

app/schemas/item.py

+from typing import Optional
+from datetime import datetime
+from sqlmodel import SQLModel
+
+class ItemBase(SQLModel):
+    title: str
+    description: Optional[str] = None
+    is_completed: bool = False
+
+class ItemCreate(ItemBase):
+    pass
+
+class ItemUpdate(SQLModel):
+    title: Optional[str] = None
+    description: Optional[str] = None
+    is_completed: Optional[bool] = None
+
+class ItemResponse(ItemBase):
+    id: int
+    created_at: datetime
+    updated_at: datetime

app/services/base.py

+from typing import Generic, TypeVar, List, Optional
+from app.repositories.base import BaseRepository
+from sqlmodel import SQLModel
+
+ModelType = TypeVar("ModelType", bound=SQLModel)
+RepositoryType = TypeVar("RepositoryType", bound=BaseRepository)
+
+class BaseService(Generic[ModelType, RepositoryType]):
+    """
+    A generic service class implementing standard business logic for CRUD operations.
+
+    This class acts as a bridge between the API (Controller) and the Data Access Layer (Repository).
+    It can be extended to add specific business rules.
+
+    Attributes:
+        repository (RepositoryType): The repository instance used for data access.
+    """
+    def __init__(self, repository: RepositoryType):
+        """
+        Args:
+            repository (RepositoryType): The repository instance.
+        """
+        self.repository = repository
+
+    def get_all(self) -> List[ModelType]:
+        """Returns all records from the repository."""
+        return self.repository.get_all()
+
+    def get_by_id(self, id: int) -> Optional[ModelType]:
+        """Returns a record by ID."""
+        return self.repository.get_by_id(id)
+
+    def create(self, obj_in: ModelType) -> ModelType:
+        """Creates a new record."""
+        return self.repository.create(obj_in)
+
+    def update(self, id: int, obj_in_data: dict) -> Optional[ModelType]:
+        """
+        Updates a record by ID.
+
+        Returns:
+            Optional[ModelType]: The updated object, or None if not found.
+        """
+        db_obj = self.repository.get_by_id(id)
+        if not db_obj:
+            return None
+        return self.repository.update(db_obj, obj_in_data)
+
+    def delete(self, id: int) -> Optional[ModelType]:
+        """
+        Deletes a record by ID.
+
+        Returns:
+            Optional[ModelType]: The deleted object, or None if not found.
+        """
+        db_obj = self.repository.get_by_id(id)
+        if not db_obj:
+            return None
+        return self.repository.delete(db_obj)

app/services/item.py

+from app.models.item import Item
+from app.repositories.item import ItemRepository
+from app.services.base import BaseService
+
+class ItemService(BaseService[Item, ItemRepository]):
+    def __init__(self, repository: ItemRepository):
+        super().__init__(repository)

verify.py

+import requests
+import sys
+
+BASE_URL = "http://127.0.0.1:8000"
+
+def test_crud():
+    print(f"Testing against {BASE_URL}")
+
+    # 1. Create
+    payload = {"title": "Test Item", "description": "Verification Item"}
+    response = requests.post(f"{BASE_URL}/items/", json=payload)
+    if response.status_code != 200:
+        print(f"Create failed: {response.text}")
+        sys.exit(1)
+    data = response.json()
+    item_id = data["id"]
+    print(f"Created Item: {data}")
+
+    # 2. Key Get
+    response = requests.get(f"{BASE_URL}/items/{item_id}")
+    if response.status_code != 200:
+        print(f"Get failed: {response.text}")
+        sys.exit(1)
+    print(f"Got Item: {response.json()}")
+
+    # 3. List
+    response = requests.get(f"{BASE_URL}/items/")
+    if response.status_code != 200:
+        print(f"List failed: {response.text}")
+        sys.exit(1)
+    print(f"List Items: {len(response.json())} items")
+
+    # 4. Update
+    payload = {"title": "Updated Item", "is_completed": True}
+    response = requests.put(f"{BASE_URL}/items/{item_id}", json=payload)
+    if response.status_code != 200:
+        print(f"Update failed: {response.text}")
+        sys.exit(1)
+    print(f"Updated Item: {response.json()}")
+
+    # 5. Delete
+    response = requests.delete(f"{BASE_URL}/items/{item_id}")
+    if response.status_code != 200:
+        print(f"Delete failed: {response.text}")
+        sys.exit(1)
+    print(f"Deleted Item: {response.json()}")
+
+    # 6. Verify Delete
+    response = requests.get(f"{BASE_URL}/items/{item_id}")
+    if response.status_code != 404:
+        print(f"Item should be deleted but got: {response.status_code}")
+        sys.exit(1)
+    print("Verification Successful!")
+
+if __name__ == "__main__":
+    try:
+        test_crud()
+    except Exception as e:
+        print(f"Error: {e}")
+        sys.exit(1)