1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

# main.py
import sqlite3
from contextlib import asynccontextmanager
from typing import Generator, List

from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, EmailStr

DATABASE_URL = "contacts.db"

# ---------- Model ----------
# This defines the shape of our data
class Contact(BaseModel):
    id: int | None = None
    first_name: str
    last_name: str
    email: EmailStr

# ---------- Lifespan + App Initialization ----------
# This runs on app startup
@asynccontextmanager
async def lifespan(app: FastAPI):
    conn = sqlite3.connect(DATABASE_URL)
    conn.row_factory = sqlite3.Row
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS contacts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            first_name TEXT NOT NULL,
            last_name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL
        )
        """
    )
    conn.commit()
    conn.close()
    yield
    # No shutdown logic needed for this demo

app = FastAPI(title="Contact Manager", lifespan=lifespan)

# ---------- Dependency for DB Connection ----------
# This gives us a database connection for each request
def get_db() -> Generator[sqlite3.Connection, None, None]:
    conn = sqlite3.connect(DATABASE_URL)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()

# ---------- Routes (The "API" part) ----------
@app.get("/contacts", response_model=List[Contact])
def list_contacts(conn: sqlite3.Connection = Depends(get_db)) -> List[Contact]:
    rows = conn.execute("SELECT * FROM contacts").fetchall()
    return [Contact(**dict(row)) for row in rows]

@app.get("/contacts-initials", response_model=List[str])
def list_full_name_initials(conn: sqlite3.Connection = Depends(get_db)) -> List[str]:
    # this is just an example to show that not all operations are simple CRUD operations
    rows = conn.execute("SELECT * FROM contacts").fetchall()
    contacts = [Contact(**dict(row)) for row in rows]
    
    name_initials: List[str] = [
        f"{contact.first_name[0]}. {contact.last_name[0]}."
        for contact in contacts
        if contact.first_name and contact.last_name
    ]
    return name_initials

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# repository/interface.py
from abc import ABC, abstractmethod
from typing import List
from models import Contact # (Assuming models.py exists)

class IContactRepository(ABC):
    """
    This is the interface (or "port") for our
    contact data storage.
    """
    @abstractmethod
    def get_all(self) -> List[Contact]:
        pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

# repository/sqlite.py
import sqlite3
from typing import List
from models import Contact
from repository.interface import IContactRepository

DATABASE_URL = "contacts.db" # Assume this is configured

class SqliteContactRepository(IContactRepository):
    def get_all(self) -> List[Contact]:
        # All the SQLite-specific code is now hidden in here
        with sqlite3.connect(DATABASE_URL) as conn:
            conn.row_factory = sqlite3.Row
            rows = conn.execute("SELECT * FROM contacts").fetchall()
            return [Contact(**dict(row)) for row in rows]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# repository/mongodb.py
import os
from typing import List
from models import Contact
from repository.interface import IContactRepository
from pymongo import MongoClient

class MongoContactRepository(IContactRepository):
    def get_all(self) -> List[Contact]:
        # All the PyMongo-specific code would go here
        client = MongoClient(os.getenv("MONGO_URI"))
        db = client[os.getenv("DB_NAME")]
        return [Contact(**data) for data in db.contacts.find({}, {"_id": 0})]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

# repository/factory.py
import os
from typing import Dict, Type
from repository.interface import IContactRepository
from repository.sqlite import SqliteContactRepository
from repository.mongodb import MongoContactRepository

repositories_map: Dict[str, Type[IContactRepository]] = {
    "sqlite": SqliteContactRepository,
    "mongodb": MongoContactRepository,
}

def get_repository() -> IContactRepository:
    """
    Dependency factory to get the correct
    repository based on environment.
    """
    db_type = os.environ.get("DB_TYPE", "sqlite")
    _repo_cls = repositories_map[db_type]
    _repo_instance = _repo_cls()
    return _repo_instance

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# main.py
from fastapi import Depends, FastAPI, HTTPException
from models import Contact
from repository.factory import get_repository
from repository.interface import IContactRepository

# ... (app setup, lifespan, etc.)

@app.get("/contacts", response_model=List[Contact])
def list_contacts(
    repo: IContactRepository = Depends(get_repository)
) -> List[Contact]:
    return repo.get_all()

@app.get("/contacts-initials", response_model=List[str])
def list_full_name_initials(
    repo: IContactRepository = Depends(get_repository)
) -> List[str]:
    contacts = repo.get_all()
    name_initials: List[str] = [
        f"{contact.first_name[0]}. {contact.last_name[0]}."
        for contact in contacts
        if contact.first_name and contact.last_name
    ]
    return name_initials

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

@app.get("/contacts-initials", response_model=List[str])
def list_full_name_initials(
    repo: IContactRepository = Depends(get_repository)
) -> List[str]:
    
    # --- This is our Business Logic ---
    contacts = repo.get_all()
    name_initials: List[str] = [
        f"{contact.first_name[0]}. {contact.last_name[0]}."
        for contact in contacts
        if contact.first_name and contact.last_name
    ]
    # --- End Business Logic ---
    
    return name_initials

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

# lambda_handler.py
from typing import List, Dict, Any
from repository.factory import get_repository

def handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    repo = get_repository()
    
    # --- COPY-PASTED Business Logic ---
    contacts = repo.get_all()
    name_initials: List[str] = [
        f"{contact.first_name[0]}. {contact.last_name[0]}."
        for contact in contacts
        if contact.first_name and contact.last_name
    ]
    # --- End Business Logic ---
    
    return {"statusCode": 200, "body": name_initials}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

# services/contact_service.py
from typing import List
from models import Contact
from repository.interface import IContactRepository

class ContactService:
    def __init__(self, repo: IContactRepository):
        # It depends on the interface, not a concrete type
        self.repo = repo

    def get_all_contacts(self) -> List[Contact]:
        return self.repo.get_all()

    def get_contact_initials(self) -> List[str]:
        # All our business logic is now in *one* place.
        contacts = self.repo.get_all()
        name_initials: List[str] = [
            f"{contact.first_name[0]}. {contact.last_name[0]}."
            for contact in contacts
            if contact.first_name and contact.last_name
        ]
        return name_initials

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# main.py
from fastapi import Depends, FastAPI, HTTPException
from models import Contact
from services.contact_service import ContactService
from repository.factory import get_repository # Added
from repository.interface import IContactRepository # Added

app = FastAPI() # ... etc.

@app.get("/contacts", response_model=List[Contact])
def list_contacts(
    repo: IContactRepository = Depends(get_repository)
) -> List[Contact]:
    # for simplicity we didn't create a factory for the service
    # the service layer is where your business rules live
    service = ContactService(repo=repo)
    return service.get_all_contacts()

@app.get("/contacts-initials", response_model=List[str])
def list_full_name_initials(
    repo: IContactRepository = Depends(get_repository)
) -> List[str]:
    # for simplicity we didn't create a factory for the service
    service = ContactService(repo=repo)
    # All logic is hidden inside the service
    return service.get_contact_initials()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# lambda_handler.py
from repository.factory import get_repository
from services.contact_service import ContactService
from typing import Dict, Any # Added

def handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    # Manually create the service (or use a DI container)
    repo = get_repository()
    service = ContactService(repo=repo)
    
    # Call the *same* business logic method
    initials = service.get_contact_initials()
    
    return {"statusCode": 200, "body": initials}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# tests/test_service.py
class MockContactRepository(IContactRepository):
    def get_all(self) -> List[Contact]:
        return [
            Contact(id=1, first_name="Test", last_name="User", email="test@user.com")
        ]
    # ... other methods


def test_initials_logic() -> None:
    mock_repo = MockContactRepository()
    service = ContactService(repo=mock_repo)

    initials = service.get_contact_initials()

    assert initials == ["T. U."]