store apis implemented

Author: lordlabakdas

api/helpers/inventory_helper.py

@@ -9,6 +9,7 @@
 
 logger = logging.getLogger()
 
+
 def to_dict(self):
     return {
         "name": self.name,
@@ -18,6 +19,7 @@ def to_dict(self):
         "notes": self.notes,
     }
 
+
 class InventoryHelper(object):
     def __init__(self, user: User) -> None:
         self.user = user

api/store_api.py

@@ -0,0 +1,102 @@
+from typing import List
+from uuid import UUID
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel
+
+from helpers.jwt import auth_required
+from models.entity.user_entity import User
+from models.service.store_service import StoreService
+
+store_apis = APIRouter(prefix="/stores", tags=["stores"])
+
+
+class StoreCreate(BaseModel):
+    name: str
+    address: str
+    phone: str
+    email: str
+    website: str = None
+
+
+class StoreUpdate(BaseModel):
+    name: str = None
+    address: str = None
+    phone: str = None
+    email: str = None
+    website: str = None
+
+
+@store_apis.post("/", status_code=status.HTTP_201_CREATED)
+async def create_store(
+    store_data: StoreCreate, current_user: User = Depends(auth_required)
+):
+    store_service = StoreService()
+    return store_service.create_store(user_id=current_user.id, **store_data.dict())
+
+
+@store_apis.get("/{store_id}")
+async def get_store(store_id: UUID, current_user: User = Depends(auth_required)):
+    store_service = StoreService()
+    return store_service.get_store(store_id)
+
+
+@store_apis.get("/")
+async def get_user_stores(current_user: User = Depends(auth_required)):
+    store_service = StoreService()
+    return store_service.get_stores_by_user(current_user.id)
+
+
+@store_apis.put("/{store_id}")
+async def update_store(
+    store_id: UUID, store_data: StoreUpdate, current_user: User = Depends(auth_required)
+):
+    store_service = StoreService()
+    store = store_service.get_store(store_id)
+    if store.user_id != current_user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Not authorized to update this store",
+        )
+    return store_service.update_store(store_id, **store_data.dict(exclude_unset=True))
+
+
+@store_apis.delete("/{store_id}")
+async def delete_store(store_id: UUID, current_user: User = Depends(auth_required)):
+    store_service = StoreService()
+    store = store_service.get_store(store_id)
+    if store.user_id != current_user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Not authorized to delete this store",
+        )
+    return store_service.delete_store(store_id)
+
+
+@store_apis.post("/{store_id}/deactivate")
+async def deactivate_store(store_id: UUID, current_user: User = Depends(auth_required)):
+    store_service = StoreService()
+    store = store_service.get_store(store_id)
+    if store.user_id != current_user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Not authorized to deactivate this store",
+        )
+    return store_service.deactivate_store(store_id)
+
+
+@store_apis.post("/{store_id}/activate")
+async def activate_store(store_id: UUID, current_user: User = Depends(auth_required)):
+    store_service = StoreService()
+    store = store_service.get_store(store_id)
+    if store.user_id != current_user.id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Not authorized to activate this store",
+        )
+    return store_service.activate_store(store_id)
+
+
+@store_apis.get("/search/{query}")
+async def search_stores(query: str, current_user: User = Depends(auth_required)):
+    store_service = StoreService()
+    return store_service.search_stores(query)

engine/inventory/cart.py

@@ -1,8 +1,9 @@
 from models.entity.user_entity import User
+from models.entity.store_entity import Store
 
 
 class Cart(object):
-    def __init__(self, user: User):
+    def __init__(self, user: User, store: Store):
         self.user = user
         self.items = []
 

main.py

@@ -4,10 +4,13 @@
 import uvicorn
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.openapi.utils import get_openapi
+from fastapi.security import OAuth2PasswordBearer
 from loguru import logger
 
 from api.google_login import google_login_apis
 from api.inventory_api import inventory_apis
+from api.store_api import store_apis
 from api.user_api import user_apis
 from models.db import create_db_and_tables
 
@@ -49,8 +52,38 @@ async def welcome():
 # app.include_router(firebase_api, prefix="/firebase")
 app.include_router(google_login_apis, prefix="/google")
 app.include_router(inventory_apis, prefix="/inventory")
+app.include_router(store_apis, prefix="/store")
 
 create_db_and_tables()
 
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+
+def custom_openapi():
+    if app.openapi_schema:
+        return app.openapi_schema
+    openapi_schema = get_openapi(
+        title="Your API Title",
+        version="1.0.0",
+        description="Your API Description",
+        routes=app.routes,
+    )
+    openapi_schema["components"]["securitySchemes"] = {
+        "BearerAuth": {
+            "type": "http",
+            "scheme": "bearer",
+            "bearerFormat": "JWT",
+            "description": "Enter your JWT token in the format: Bearer <token>",
+        }
+    }
+    for path in openapi_schema["paths"].values():
+        for method in path.values():
+            method.setdefault("security", []).append({"BearerAuth": []})
+    app.openapi_schema = openapi_schema
+    return app.openapi_schema
+
+
+app.openapi = custom_openapi
+
 if __name__ == "__main__":
     uvicorn.run(app, port=8000)

models/entity/cart_entity.py

@@ -5,7 +5,7 @@
 from models.entity.cart_item_entity import CartItemEntity
 
 
-class CartEntity(SQLModel, table=True   ):
+class CartEntity(SQLModel, table=True):
     id: Optional[UUID] = Field(default_factory=uuid4, primary_key=True)
     user_id: UUID = Field(foreign_key="user.id")
     items: List[CartItemEntity] = Field(default=[])
@@ -25,7 +25,6 @@ def remove_item(self, item: CartItemEntity):
         self.total_price -= item.price
         self.total_quantity -= item.quantity
 
-
     def clear(self):
         self.items = []
         self.total_price = 0.0

models/entity/cart_item_entity.py

@@ -3,12 +3,13 @@
 from uuid import UUID, uuid4
 from sqlmodel import Field, SQLModel
 
+
 class CartItemEntity(SQLModel, table=True):
     id: Optional[UUID] = Field(default_factory=uuid4, primary_key=True)
     cart_id: UUID = Field(foreign_key="cartentity.id")
     inventory_id: UUID = Field(foreign_key="inventory.id")
-    quantity: int = Field(default=1) 
+    quantity: int = Field(default=1)
     price: float = Field(default=0.0)
     created_at: datetime = Field(default_factory=datetime.utcnow)
     updated_at: datetime = Field(default_factory=datetime.utcnow)
-    notes: Optional[str] = None
+    notes: Optional[str] = None

models/entity/store_entity.py

@@ -0,0 +1,41 @@
+from datetime import datetime
+from typing import Optional
+from uuid import UUID, uuid4
+from sqlmodel import Field, Relationship, SQLModel
+
+from models.entity.user_entity import User
+
+
+class Store(SQLModel, table=True):
+    id: Optional[UUID] = Field(default_factory=uuid4, primary_key=True)
+    name: str = Field(index=True)
+    user_id: UUID = Field(foreign_key="user.id")
+    # user: User = Relationship(back_populates="stores")
+    address: str = Field(index=True)
+    phone: str = Field(index=True)
+    email: str = Field(index=True)
+    website: str = Field(index=True)
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    updated_at: datetime = Field(default_factory=datetime.utcnow)
+    is_active: bool = Field(default=True)
+
+    def __repr__(self):
+        return f"<Store(id={self.id}, name={self.name})>"
+
+    def __str__(self):
+        return self.name
+
+    def __hash__(self):
+        return hash(self.id)
+
+    def is_active(self):
+        return self.is_active
+
+    def is_inactive(self):
+        return not self.is_active
+
+    def get_email(self):
+        return self.email
+
+    def get_phone(self):
+        return self.phone

models/service/cart_service.py

@@ -1,6 +1,7 @@
 from models.entity.cart_entity import CartEntity
 from models.entity.cart_item_entity import CartItemEntity
 
+
 class CartService(object):
     def __init__(self, cart_entity: CartEntity):
         self.cart_entity = cart_entity

models/service/store_service.py

@@ -0,0 +1,102 @@
+from typing import List, Optional
+from uuid import UUID
+from datetime import datetime
+from fastapi import HTTPException, status
+
+from models.db import db_session
+from models.entity.store_entity import Store
+from models.entity.user_entity import User
+
+
+class StoreService:
+    def create_store(
+        self,
+        name: str,
+        user_id: UUID,
+        address: str,
+        phone: str,
+        email: str,
+        website: str = None,
+    ) -> Store:
+        try:
+            store = Store(
+                name=name,
+                user_id=user_id,
+                address=address,
+                phone=phone,
+                email=email,
+                website=website,
+            )
+            db_session.add(store)
+            db_session.commit()
+            db_session.refresh(store)
+            return store
+        except Exception as e:
+            db_session.rollback()
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Failed to create store: {str(e)}",
+            )
+
+    def get_store(self, store_id: UUID) -> Optional[Store]:
+        store = db_session.query(Store).filter(Store.id == store_id).first()
+        if not store:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail=f"Store with id {store_id} not found",
+            )
+        return store
+
+    def get_stores_by_user(self, user_id: UUID) -> List[Store]:
+        return db_session.query(Store).filter(Store.user_id == user_id).all()
+
+    def get_store_by_email(self, email: str) -> Optional[Store]:
+        return db_session.query(Store).filter(Store.email == email).first()
+
+    def update_store(self, store_id: UUID, **update_data) -> Store:
+        try:
+            store = self.get_store(store_id)
+            for key, value in update_data.items():
+                if hasattr(store, key):
+                    setattr(store, key, value)
+            store.updated_at = datetime.utcnow()
+            db_session.commit()
+            db_session.refresh(store)
+            return store
+        except Exception as e:
+            db_session.rollback()
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Failed to update store: {str(e)}",
+            )
+
+    def delete_store(self, store_id: UUID) -> bool:
+        try:
+            store = self.get_store(store_id)
+            db_session.delete(store)
+            db_session.commit()
+            return True
+        except Exception as e:
+            db_session.rollback()
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Failed to delete store: {str(e)}",
+            )
+
+    def deactivate_store(self, store_id: UUID) -> Store:
+        return self.update_store(store_id, is_active=False)
+
+    def activate_store(self, store_id: UUID) -> Store:
+        return self.update_store(store_id, is_active=True)
+
+    def search_stores(self, query: str) -> List[Store]:
+        return (
+            db_session.query(Store)
+            .filter(
+                (Store.name.ilike(f"%{query}%"))
+                | (Store.email.ilike(f"%{query}%"))
+                | (Store.phone.ilike(f"%{query}%"))
+                | (Store.address.ilike(f"%{query}%"))
+            )
+            .all()
+        )