WIP: initial fastapi endpoint implementation

Client endpoint - GET & POST implemented

.env

@@ -1,9 +1,8 @@
-DATABASE_URL=mysql+mysqldb://admin:Avatarme1@localhost:3306/CMT
+# Domain
+# Will be set to the prod domain with an env var on deployment used by Traefik
+# to transmit traffic and acquire TLS certs
+ENVIRONMENT=local
+PROJECT_NAME="CMT"
 
-# mysql
+# DB URI
-MYSQL_ROOT_PASSWORD=%40Avatarme1
+DATABASE_URI=mysql+mysqldb://admin:Avatarme1@localhost:3306/CMT
-MYSQL_SERVER=localhost
-MYSQL_PORT=3306
-MYSQL_DB=CMT
-MYSQL_USER=admin
-MYSQL_PASSWORD=Avatarme1

backend/README.md

@@ -0,0 +1,9 @@
+# CMT Backend
+## Usage
+### API
+### Alembic
+```bash
+cd backend
+alembic revision --autogenerate -m "Header message"
+alembic upgrade head
+```

backend/app/alembic/env.py

@@ -7,7 +7,7 @@ from alembic import context
 import os
 from dotenv import load_dotenv
 from sqlmodel import SQLModel
-from app.models import Client, Supplier, Product, Payment, Credit
+from app.schemas.models import Client, Supplier, Product, Payment, Credit
 
 load_dotenv()
 # this is the Alembic Config object, which provides
@@ -24,7 +24,7 @@ if config.config_file_name is not None:
 # from myapp import mymodel
 # target_metadata = mymodel.Base.metadata
 target_metadata = SQLModel.metadata
-url = os.getenv("DATABASE_URL")
+url = os.getenv("DATABASE_URI")
 
 # other values from the config, defined by the needs of env.py,
 # can be acquired:

backend/app/alembic/versions/e8c4300db3cb_checking_for_updates.py

@@ -0,0 +1,32 @@
+"""Checking for updates
+
+Revision ID: e8c4300db3cb
+Revises: bfb086d8d500
+Create Date: 2025-06-08 19:06:55.200977
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'e8c4300db3cb'
+down_revision: Union[str, None] = 'bfb086d8d500'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    pass
+    # ### end Alembic commands ###

backend/app/api/api.py

@@ -1,7 +0,0 @@
-"""
-API Home
-"""
-from fastapi import APIRouter
-
-
-api_router = APIRouter()

backend/app/api/clients/endpoints.py

@@ -0,0 +1,31 @@
+"""
+The Client table
+"""
+from fastapi import APIRouter, HTTPException 
+from sqlmodel import Session, select
+from app.api.deps import SessionDep, exists
+from app.schemas.models import Client
+from app.schemas.schemas import ClientCreate
+from typing import Sequence
+
+router = APIRouter(prefix="/clients", tags=["clients"])
+
+
+@router.get("/", response_model=Client)
+def fetch_clients(client: Client, session: SessionDep) -> Sequence[Client]:
+    """Fetch client list
+    """
+    clients = session.exec(select(Client)).all()
+    return clients
+
+
+@router.post("/", response_model=ClientCreate)
+def create_client(client_data: ClientCreate, session: SessionDep) -> Client:
+    existing = exists(session, Client, tin_number=client_data.tin_number)
+    if existing:
+        raise HTTPException(status_code=400, detail="Client with this tin number already exists")
+    client = Client.model_validate(client_data)
+    session.add(client)
+    session.commit()
+    session.refresh(client)
+    return client

backend/app/api/deps.py

@@ -0,0 +1,30 @@
+"""
+"""
+from fastapi import Depends
+from sqlmodel import Session, SQLModel, select
+from app.core.db import get_session
+from app.schemas.models import Client
+from typing import Type, Optional, Annotated
+
+
+SessionDep = Annotated[Session, Depends(get_session)]
+
+
+def exists(session: Session, model: Type[SQLModel], **filters) -> Optional[bool]:
+    """
+    Checks if a request exists in the given model using any filters.
+
+    Example:
+        exists(session, Client, phone="0781232465", tax_number="TIN123")
+    """
+    if not filters:
+        raise ValueError("At least one filter must be provided")
+
+    stmt = select(model)
+    for field, value in filters.items():
+        if not hasattr(model, field):
+            raise ValueError(f"Invalid filter field: {field}")
+        stmt = stmt.where(getattr(model, field) == value)
+
+    result = session.exec(stmt).first()
+    return result is not None

backend/app/api/main.py

@@ -0,0 +1,11 @@
+"""
+API Home
+"""
+from fastapi import APIRouter
+from app.core.config import settings
+
+
+api_router = APIRouter()
+
+#if settings.environment == "local":
+#    api_router.include_router()

backend/app/config.py

@@ -1,41 +0,0 @@
-import secrets
-import warnings
-from typing import Annotated, Any, Literal
-from pydantic import (
-    MySQLDsn
-)
-from pydantic_core import MultiHostUrl
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-class Settings(BaseSettings):
-    """
-    """
-    model_config = SettingsConfigDict(
-        # One level above ./backend
-        env_file='../.env',
-        env_ignore_empty=True,
-        extra='ignore'
-    )
-    SECRET_KEY: str = secrets.token_urlsafe(32)
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8 # 8 days
-
-    MYSQL_SERVER: str
-    MYSQL_PORT: int = 3306
-    MYSQL_USER: str
-    MYSQL_PASSWORD: str = ""
-    MYSQL_DB: str = ""
-
-    @computed_field # type: ignore[prop-decorator]
-    @property
-    def SQLALCHEMY_DATABASE_URI(self) -> MySQLDsn:
-        return MultiHostUrl.build(
-            scheme="mysql+mysqldb",
-            username=self.MYSQL_USER,
-            password=self.MYSQL_PASSWORD,
-            host=self.MYSQL_SERVER,
-            port=self.MYSQL_PORT,
-            path=self.MYSQL_DB
-        ) # type: ignore
-
-settings = Settings() # type: ignore

backend/app/core/config.py

@@ -0,0 +1,22 @@
+from pydantic import (
+    MySQLDsn
+)
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    """
+    """
+    database_uri: MySQLDsn
+    environment: str
+    project_name: str
+
+    model_config = SettingsConfigDict(
+        # One level above ./backend
+        env_file='../.env',
+        env_ignore_empty=True,
+        extra='ignore'
+    )
+    api_v1_str: str = "/api/v1"
+
+settings = Settings() # type: ignore

backend/app/core/db.py

@@ -0,0 +1,11 @@
+from sqlmodel import Session, create_engine
+from app.core.config import settings
+
+engine = create_engine(str(settings.database_uri))
+
+
+def get_session():
+    """main interface to interact with db
+    """
+    with Session(engine) as session:
+        yield session

backend/app/crud/client.py

@@ -1,15 +0,0 @@
-"""
-The Client table
-"""
-from fastapi import APIRouter, HTTPException
-from sqlmodel import func, select
-from app.models import Client, Supplier, Product, Payment, Credit
-from typing import Any
-
-router = APIRouter(prefix="/client", tags=["items"])
-
-
-@router.get("/", response_model=Client)
-def read_clients(
-    session: SessionDep,
-)

backend/app/crud/product.py

@@ -1,3 +0,0 @@
-"""
-TODO: when Credit.purchase_price is updated, update Product.purchase_price
-"""

backend/app/db.py

@@ -1,9 +0,0 @@
-from sqlmodel import Session, create_engine, select
-from app.config import settings
-from app.models import Client, Supplier
-
-engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))
-
-
-def init_db(session: Session) -> None:
-    """"""

backend/app/main.py

@@ -4,17 +4,22 @@
 NOTE:
 - 
 """
-from app.config import settings
+from app.core.config import settings
 from typing import Union
 from fastapi import FastAPI
+from app.api.clients.endpoints import router as clients_router
+
 
 app = FastAPI(
-    title=settings.PROJECT_NAME,
+    title=settings.project_name,
-    openapi_url=f"{settings.API_V1_STR}/openapi.json"
+    openapi_url=f"{settings.api_v1_str}/openapi.json"
 )
 
+
 @app.get("/")
 def read_root():
     """
     """
     return {"Hello": "World"}
+
+app.include_router(clients_router, tags=["clients"])

backend/app/schemas/schemas.py

@@ -0,0 +1,7 @@
+from sqlmodel import SQLModel
+
+
+class ClientCreate(SQLModel):
+    tin_number: int
+    names: str
+    phone_number: str

backend/app/test.py

@@ -0,0 +1,37 @@
+import pytest
+from sqlmodel import SQLModel, Session, create_engine
+
+from app.core.config import settings
+from app.api.deps import SessionDep
+from app.schemas.schemas import ClientCreate
+from app.api.clients.endpoints import create_client
+from fastapi import HTTPException
+
+
+# Fixture for a temporary in-memory database session
+@pytest.fixture
+def session():
+    engine = create_engine(str(settings.database_uri), echo=False)
+    SQLModel.metadata.create_all(engine)
+    with Session(engine) as session:
+        yield session
+
+
+def test_create_client_success(session: SessionDep):
+    client_data = ClientCreate(tin_number=781410046, names="Lin", phone_number="0781410046")
+    client = create_client(client_data, session)
+
+    assert client.id is not None
+    assert client.names == "Lin"
+    assert client.tin_number == 781410046
+    assert client.phone_number == "0781410046"
+
+
+def test_create_client_duplicate_name(session):
+    client_data = ClientCreate(tin_number=781410045, names="John Doe", phone_number="0781410045")
+
+    # Try to create another client with the same name
+    with pytest.raises(HTTPException) as exc_info:
+        create_client(client_data, session)
+    assert exc_info.value.status_code == 400
+    assert "already exists" in exc_info.value.detail