add rabbitmq broker for user registration events

Author: lordlabakdas

api/firebase_api.py

@@ -4,12 +4,14 @@
 from firebase_admin import auth as firebase_auth
 from google.oauth2 import id_token
 
-from api.validators.user_validation import (ChangePasswordPayload,
-                                            ChangePasswordResponse,
-                                            FirebaseLoginPayload,
-                                            FirebaseLoginResponse,
-                                            FirebaseRegistrationPayload,
-                                            FirebaseRegistrationResponse)
+from api.validators.user_validation import (
+    ChangePasswordPayload,
+    ChangePasswordResponse,
+    FirebaseLoginPayload,
+    FirebaseLoginResponse,
+    FirebaseRegistrationPayload,
+    FirebaseRegistrationResponse,
+)
 from models.service.user_service import User
 
 logger = logging.getLogger("groceror")

api/inventory_api.py

@@ -4,9 +4,11 @@
 from fastapi import APIRouter, Depends, HTTPException, status
 
 from api.helpers.inventory_helper import InventoryHelper
-from api.validators.inventory_validation import (AddInventoryPayload,
-                                                 AddInventoryResponse,
-                                                 StoreInventoryResponse)
+from api.validators.inventory_validation import (
+    AddInventoryPayload,
+    AddInventoryResponse,
+    StoreInventoryResponse,
+)
 from helpers.jwt import auth_required
 from models.entity.user_entity import User
 

api/user_api.py

@@ -4,13 +4,19 @@
 from loguru import logger
 
 from api.helpers import auth_helper
-from api.validators.user_validation import (ChangePasswordPayload,
-                                            ChangePasswordResponse,
-                                            LoginPayload, LoginResponse,
-                                            RegistrationPayload,
-                                            RegistrationResponse)
+from api.validators.user_validation import (
+    ChangePasswordPayload,
+    ChangePasswordResponse,
+    LoginPayload,
+    LoginResponse,
+    RegistrationPayload,
+    RegistrationResponse,
+)
 from config import JWTConfig
+from engine import publisher
 from helpers.jwt import JWT
+import pika
+import json
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
@@ -54,6 +60,9 @@ async def register(registration_payload: RegistrationPayload):
             detail="Issue with registering user",
         )
     else:
+        publisher.publish_message(
+            event="user_registered", routing_key="email_queue", new_user=new_user
+        )
         return {"id": new_user.id}
 
 

engine/publisher.py

@@ -0,0 +1,41 @@
+import json
+import logging
+import pika
+
+from models.entity.user_entity import User
+
+logger = logging.getLogger(__name__)
+
+def publish_message(event: str, routing_key: str, new_user: User):
+    # Publish registration event to email service
+        try:
+            connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
+            channel = connection.channel()
+            
+            # Ensure exchange exists
+            channel.queue_declare(queue='email_queue', durable=True)
+            
+            # Create message payload
+            message = {
+                "event": event,
+                "user_id": str(new_user.id),
+                "email": new_user.email,
+                "name": new_user.name
+            }
+            
+            # Publish message
+            channel.basic_publish(
+                exchange='',
+                routing_key=routing_key,
+                body=json.dumps(message),
+                properties=pika.BasicProperties(
+                    delivery_mode=2,  # make message persistent
+                    content_type='application/json'
+                )
+            )
+            
+            connection.close()
+            
+        except Exception as e:
+            logger.error(f"Failed to publish registration event: {str(e)}")
+            # Don't raise exception - registration succeeded even if notification fails

models/entity/store_entity.py

@@ -1,5 +1,5 @@
 from datetime import datetime
-from typing import Optional
+from typing import List, Optional
 from uuid import UUID, uuid4
 
 from sqlmodel import Field, Relationship, SQLModel
@@ -23,6 +23,8 @@ class Store(SQLModel, table=True):
     is_active: bool = Field(default=True)
     latitude: float = Field(default=None)
     longitude: float = Field(default=None)
+    # user: User = Relationship(back_populates="store")
+    # inventory: List["Inventory"] = Relationship(back_populates="store")
 
     def __repr__(self):
         return f"<Store(id={self.id}, name={self.name})>"

models/entity/user_entity.py

@@ -1,9 +1,9 @@
 import uuid
 from datetime import datetime
 from enum import Enum
-from typing import Optional
+from typing import List, Optional
 
-from sqlmodel import Field, SQLModel
+from sqlmodel import Field, Relationship, SQLModel
 
 
 class UserType(str, Enum):
@@ -22,3 +22,4 @@ class User(SQLModel, table=True):
     updated_at: datetime = Field(default_factory=datetime.utcnow, nullable=False)
     location: Optional[str] = None
     # inventory: List["Inventory"] = Relationship(back_populates="user")
+    # store: Optional["Store"] = Relationship(back_populates="user")

models/service/inventory_service.py

@@ -8,7 +8,7 @@ class InventoryService:
     def __init__(self, store_id: str):
         self.store_id = store_id
 
-    def get_inventory(self) -> List[Inventory]:
+    def get_inventory_for_store(self) -> List[Inventory]:
         return (
             db_session.query(Inventory)
             .filter(Inventory.store_id == self.store_id)
@@ -24,5 +24,23 @@ def add_inventory(
         db_session.add(inventory)
         db_session.commit()
 
+    def delete_inventory(self, name: str):
+        store_inventory = (
+            db_session.query(Inventory)
+            .filter(Inventory.name == name, Inventory.store_id == self.store_id)
+            .first()
+        )
+        db_session.delete(store_inventory)
+        db_session.commit()
+
     def update_inventory(self, inventory: Inventory):
+        store_inventory = (
+            db_session.query(Inventory)
+            .filter(
+                Inventory.name == inventory.name, Inventory.store_id == self.store_id
+            )
+            .first()
+        )
+        store_inventory.quantity = inventory.quantity
+        store_inventory.price = inventory.price
         db_session.commit()

models/service/store_service.py

@@ -182,3 +182,17 @@ def find_nearby_stores(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail=f"Error finding nearby stores: {str(e)}",
             )
+
+    def get_store_user_specs(self, store_id: UUID) -> dict:
+        store = self.get_store(store_id)
+        user = db_session.query(User).filter(User.id == store.user_id).first()
+        return {
+            "name": user.name,
+            "email": user.email,
+            "phone": user.phone,
+            "address": user.address,
+        }
+
+    def get_store_email(self, store_id: UUID) -> str:
+        store = self.get_store(store_id)
+        return db_session.query(User).filter(User.id == store.user_id).first().email

models/service/user_service.py

@@ -5,6 +5,7 @@
 import sqlalchemy
 
 from helpers.exceptions import GrocerorError
+
 # from helpers.jwt import JWT
 from models.db import db_session
 from models.entity.user_entity import User