inventory-service initHEADmain

12 files changed, 415 insertions, 0 deletions

diff --git a/inventory-service/app/api/deps.py b/inventory-service/app/api/deps.py
new file mode 100644
index 0000000..d8dd349
--- /dev/null
+++ b/inventory-service/app/api/deps.py
@@ -0,0 +1,10 @@
+from typing import Generator
+from app.core.database import SessionLocal
+
+
+def get_db() -> Generator:
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
diff --git a/inventory-service/app/api/v1/api.py b/inventory-service/app/api/v1/api.py
new file mode 100644
index 0000000..b5d56d3
--- /dev/null
+++ b/inventory-service/app/api/v1/api.py
@@ -0,0 +1,9 @@
+from fastapi import APIRouter
+from app.api.v1.endpoints import products, stock_movements
+
+api_router = APIRouter()
+
+api_router.include_router(products.router, prefix="/products", tags=["products"])
+api_router.include_router(
+    stock_movements.router, prefix="/stock-movements", tags=["stock-movements"]
+)
diff --git a/inventory-service/app/api/v1/endpoints/products.py b/inventory-service/app/api/v1/endpoints/products.py
new file mode 100644
index 0000000..2e5a2b5
--- /dev/null
+++ b/inventory-service/app/api/v1/endpoints/products.py
@@ -0,0 +1,54 @@
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.orm import Session
+from typing import List
+
+from app.api import deps
+from app.schemas.inventory import ProductCreate, ProductUpdate, ProductOut
+
+
+router = APIRouter()
+
+
[email protected]("/", response_model=ProductOut, status_code=status.HTTP_201_CREATED)
+def create_product(product_in: ProductCreate, db: Session = Depends(deps.get_db)):
+    """
+    Register a new product with an initial stock level.
+    """
+    pass  # TODO: implement
+
+
[email protected]("/", response_model=List[ProductOut])
+def list_products(skip: int = 0, limit: int = 100, db: Session = Depends(deps.get_db)):
+    """
+    Return a paginated list of all products and their current stock levels.
+    """
+    pass  # TODO: implement
+
+
[email protected]("/{product_id}", response_model=ProductOut)
+def get_product(product_id: int, db: Session = Depends(deps.get_db)):
+    """
+    Return a single product by its database ID.
+    """
+    pass  # TODO: implement
+
+
[email protected]("/{product_id}", response_model=ProductOut)
+def update_product(
+    product_id: int,
+    product_in: ProductUpdate,
+    db: Session = Depends(deps.get_db),
+):
+    """
+    Partially update a product (name or stock level).
+    Useful for manual stock adjustments by warehouse staff.
+    """
+    pass  # TODO: implement
+
+
[email protected]("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_product(product_id: int, db: Session = Depends(deps.get_db)):
+    """
+    Remove a product from the inventory.
+    """
+    pass  # TODO: implement
diff --git a/inventory-service/app/api/v1/endpoints/stock_movements.py b/inventory-service/app/api/v1/endpoints/stock_movements.py
new file mode 100644
index 0000000..f70396f
--- /dev/null
+++ b/inventory-service/app/api/v1/endpoints/stock_movements.py
@@ -0,0 +1,47 @@
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.orm import Session
+from typing import List
+
+from app.api import deps
+from app.schemas.inventory import StockMovementOut
+
+
+router = APIRouter()
+
+
[email protected]("/", response_model=List[StockMovementOut])
+def list_stock_movements(
+    skip: int = 0,
+    limit: int = 100,
+    db: Session = Depends(deps.get_db),
+):
+    """
+    Return a paginated list of all stock movements across all products.
+    Useful for auditing the full history of reservations and releases.
+    """
+    pass  # TODO: implement
+
+
[email protected]("/product/{product_id}", response_model=List[StockMovementOut])
+def list_movements_for_product(
+    product_id: int,
+    skip: int = 0,
+    limit: int = 100,
+    db: Session = Depends(deps.get_db),
+):
+    """
+    Return all stock movements for a specific product.
+    """
+    pass  # TODO: implement
+
+
[email protected]("/order/{order_id}", response_model=List[StockMovementOut])
+def list_movements_for_order(
+    order_id: int,
+    db: Session = Depends(deps.get_db),
+):
+    """
+    Return all stock movements associated with a specific order.
+    Lets you see exactly what was reserved/released for a given order.
+    """
+    pass  # TODO: implement
diff --git a/inventory-service/app/core/config.py b/inventory-service/app/core/config.py
new file mode 100644
index 0000000..df5526d
--- /dev/null
+++ b/inventory-service/app/core/config.py
@@ -0,0 +1,19 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    PROJECT_NAME: str = "Inventory Service"
+
+    DATABASE_URL: str
+
+    KAFKA_BOOTSTRAP_SERVERS: str = "localhost:9092"
+    KAFKA_CONSUMER_GROUP_ID: str = "inventory-service-group"
+    KAFKA_ORDERS_TOPIC: str = "orders"
+    KAFKA_INVENTORY_TOPIC: str = "inventory"
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+
+settings = Settings()
diff --git a/inventory-service/app/core/database.py b/inventory-service/app/core/database.py
new file mode 100644
index 0000000..3d8f1c9
--- /dev/null
+++ b/inventory-service/app/core/database.py
@@ -0,0 +1,8 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import declarative_base, sessionmaker
+from app.core.config import settings
+
+engine = create_engine(settings.DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
diff --git a/inventory-service/app/init_db.py b/inventory-service/app/init_db.py
new file mode 100644
index 0000000..0d6370e
--- /dev/null
+++ b/inventory-service/app/init_db.py
@@ -0,0 +1,13 @@
+from app.core.database import engine, Base
+
+from app.models.inventory import Product, StockMovement, ProcessedEvent  # noqa: F401
+
+
+def init_database():
+    print("Creating inventory database tables...")
+    Base.metadata.create_all(bind=engine)
+    print("Inventory database initialized successfully! 🎉")
+
+
+if __name__ == "__main__":
+    init_database()
diff --git a/inventory-service/app/kafka_consumer.py b/inventory-service/app/kafka_consumer.py
new file mode 100644
index 0000000..9fda00d
--- /dev/null
+++ b/inventory-service/app/kafka_consumer.py
@@ -0,0 +1,36 @@
+from confluent_kafka import Consumer, KafkaException
+from app.core.config import settings
+from app.core.database import SessionLocal
+
+
+def _build_consumer() -> Consumer:
+    conf = {
+        "bootstrap.servers": settings.KAFKA_BOOTSTRAP_SERVERS,
+        "group.id": settings.KAFKA_CONSUMER_GROUP_ID,
+        "auto.offset.reset": "earliest",
+        "enable.auto.commit": False,
+    }
+    consumer = Consumer(conf)
+    consumer.subscribe([settings.KAFKA_ORDERS_TOPIC])
+    return consumer
+
+
+def handle_order_created(message_value: dict, db) -> None:
+
+    pass  # TODO: implement
+
+
+def start_consumer_loop() -> None:
+
+    consumer = _build_consumer()
+    db = SessionLocal()
+
+    try:
+        while True:
+            # TODO: poll, decode, call handle_order_created, commit offset
+            pass
+    except KafkaException as e:
+        print(f"[Kafka Consumer] Fatal error: {e}")
+    finally:
+        consumer.close()
+        db.close()
diff --git a/inventory-service/app/kafka_producer.py b/inventory-service/app/kafka_producer.py
new file mode 100644
index 0000000..ede6012
--- /dev/null
+++ b/inventory-service/app/kafka_producer.py
@@ -0,0 +1,33 @@
+import json
+from confluent_kafka import Producer
+from app.core.config import settings
+
+_conf = {"bootstrap.servers": settings.KAFKA_BOOTSTRAP_SERVERS}
+_producer = Producer(_conf)
+
+
+def _delivery_report(err, msg):
+    if err is not None:
+        print(f"[Kafka] Delivery failed: {err}")
+    else:
+        print(f"[Kafka] Message delivered → {msg.topic()} [partition {msg.partition()}]")
+
+
+def send_inventory_event(event_payload: dict) -> None:
+    """
+    Publish an inventory result event to the inventory topic.
+    Called after the consumer decides to emit StockReserved or StockReservationFailed.
+
+    Args:
+        event_payload: A dict that conforms to either StockReservedEvent or
+                       StockReservationFailedEvent schema.
+    """
+    pass  # TODO: implement
+
+
+def flush_producer() -> None:
+    """
+    Block until all buffered messages are delivered.
+    Call this on application shutdown.
+    """
+    pass  # TODO: implement
diff --git a/inventory-service/app/main.py b/inventory-service/app/main.py
new file mode 100644
index 0000000..81efdd2
--- /dev/null
+++ b/inventory-service/app/main.py
@@ -0,0 +1,33 @@
+import threading
+import uvicorn
+from fastapi import FastAPI
+from app.api.v1.api import api_router
+from app.core.config import settings
+from app.kafka_consumer import start_consumer_loop
+
+app = FastAPI(title=settings.PROJECT_NAME)
+
+app.include_router(api_router, prefix="/api/v1")
+
+
[email protected]_event("startup")
+def startup_event():
+    """
+    On startup, launch the Kafka consumer in a daemon background thread
+    so the event loop and the HTTP server run concurrently.
+    """
+    consumer_thread = threading.Thread(
+        target=start_consumer_loop,
+        daemon=True,
+        name="kafka-consumer",
+    )
+    consumer_thread.start()
+
+
[email protected]("/health")
+def health_check():
+    return {"status": "working"}
+
+
+if __name__ == "__main__":
+    uvicorn.run("app.main:app", host="127.0.0.1", port=8001, reload=True)
diff --git a/inventory-service/app/models/inventory.py b/inventory-service/app/models/inventory.py
new file mode 100644
index 0000000..531d1ec
--- /dev/null
+++ b/inventory-service/app/models/inventory.py
@@ -0,0 +1,60 @@
+from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, UniqueConstraint
+from sqlalchemy.orm import relationship
+from sqlalchemy.sql import func
+from app.core.database import Base
+
+
+class Product(Base):
+    __tablename__ = "products"
+
+    id = Column(Integer, primary_key=True, index=True)
+    name = Column(String(255), nullable=False)
+    sku = Column(String(100), unique=True, nullable=False, index=True)
+    quantity_in_stock = Column(Integer, nullable=False, default=0)
+
+    created_at = Column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at = Column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )
+
+    stock_movements = relationship(
+        "StockMovement", back_populates="product", cascade="all, delete-orphan"
+    )
+
+
+class StockMovement(Base):
+    __tablename__ = "stock_movements"
+
+    id = Column(Integer, primary_key=True, index=True)
+    product_id = Column(
+        Integer,
+        ForeignKey("products.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    order_id = Column(Integer, nullable=False, index=True)
+
+    movement_type = Column(String(50), nullable=False)
+    quantity_change = Column(Integer, nullable=False)
+
+    created_at = Column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+
+    product = relationship("Product", back_populates="stock_movements")
+
+
+class ProcessedEvent(Base):
+    __tablename__ = "processed_events"
+
+    id = Column(Integer, primary_key=True, index=True)
+    order_id = Column(Integer, unique=True, nullable=False, index=True)
+
+    processed_at = Column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
diff --git a/inventory-service/app/schemas/inventory.py b/inventory-service/app/schemas/inventory.py
new file mode 100644
index 0000000..592c10a
--- /dev/null
+++ b/inventory-service/app/schemas/inventory.py
@@ -0,0 +1,93 @@
+from datetime import datetime
+from decimal import Decimal
+from pydantic import BaseModel, Field, ConfigDict
+from typing import Optional
+
+
+class ProductCreate(BaseModel):
+    name: str = Field(..., min_length=1, max_length=255)
+    sku: str = Field(
+        ...,
+        min_length=1,
+        max_length=100,
+        description="Stock-keeping unit — must be unique",
+    )
+    quantity_in_stock: int = Field(
+        ..., ge=0, description="Initial stock level, cannot be negative"
+    )
+
+
+class ProductUpdate(BaseModel):
+    name: Optional[str] = Field(None, min_length=1, max_length=255)
+    quantity_in_stock: Optional[int] = Field(None, ge=0)
+
+
+class ProductOut(BaseModel):
+    id: int
+    name: str
+    sku: str
+    quantity_in_stock: int
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class StockMovementOut(BaseModel):
+    id: int
+    product_id: int
+    order_id: int
+    movement_type: str
+    quantity_change: int
+    created_at: datetime
+
+    model_config = ConfigDict(from_attributes=True)
+
+
+class ReservedItem(BaseModel):
+    product_id: int
+    quantity: int
+
+
+class FailedItem(BaseModel):
+    product_id: int
+    quantity: int
+
+
+class StockReservedEvent(BaseModel):
+    """
+    Emitted on the 'inventory' topic when stock is successfully reserved.
+
+    Example payload:
+    {
+        "order_id": 123,
+        "status": "SUCCESS",
+        "reserved_items": [
+            {"product_id": 101, "quantity": 2},
+            {"product_id": 102, "quantity": 1}
+        ]
+    }
+    """
+
+    order_id: int
+    status: str = "SUCCESS"
+    reserved_items: list[ReservedItem]
+
+
+class StockReservationFailedEvent(BaseModel):
+    """
+    Emitted on the 'inventory' topic when stock cannot be reserved.
+
+    Example payload:
+    {
+        "order_id": 123,
+        "status": "FAILED",
+        "reason": "INSUFFICIENT_STOCK",
+        "failed_items": [{"product_id": 102, "quantity": 5}]
+    }
+    """
+
+    order_id: int
+    status: str = "FAILED"
+    reason: str = "INSUFFICIENT_STOCK"
+    failed_items: list[FailedItem]