bakery-admin/bakery-ia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Start integrating the onboarding flow with backend 17

Browse Source
This commit is contained in:
Urtzi Alfaro
2025-09-08 16:02:20 +02:00
parent 54102f7f4b
commit 201817a1be
6 changed files with 353 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
services/inventory/app/services/inventory_alert_service.py
View File

@@ -11,6 +11,7 @@ from uuid import UUID
from datetime import datetime, timedelta
import structlog
from apscheduler.triggers.cron import CronTrigger
from sqlalchemy import text
from shared.alerts.base_service import BaseAlertService, AlertServiceMixin
from shared.alerts.templates import format_item_message
@@ -23,46 +24,48 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
def setup_scheduled_checks(self):
"""Inventory-specific scheduled checks for alerts and recommendations"""
# Critical stock checks - every 5 minutes (alerts)
# SPACED SCHEDULING TO PREVENT CONCURRENT EXECUTION AND DEADLOCKS
# Critical stock checks - every 5 minutes (alerts) - Start at minute 0, 5, 10, etc.
self.scheduler.add_job(
self.check_stock_levels,
CronTrigger(minute='*/5'),
CronTrigger(minute='0,5,10,15,20,25,30,35,40,45,50,55'), # Explicit minutes
id='stock_levels',
misfire_grace_time=30,
max_instances=1
)
# Expiry checks - every 2 minutes (food safety critical, alerts)
# Expiry checks - every 2 minutes (food safety critical, alerts) - Start at minute 1, 3, 7, etc.
self.scheduler.add_job(
self.check_expiring_products,
CronTrigger(minute='*/2'),
CronTrigger(minute='1,3,7,9,11,13,17,19,21,23,27,29,31,33,37,39,41,43,47,49,51,53,57,59'), # Avoid conflicts
id='expiry_check',
misfire_grace_time=30,
max_instances=1
)
# Temperature checks - every 2 minutes (alerts)
# Temperature checks - every 5 minutes (alerts) - Start at minute 2, 12, 22, etc. (reduced frequency)
self.scheduler.add_job(
self.check_temperature_breaches,
CronTrigger(minute='*/2'),
CronTrigger(minute='2,12,22,32,42,52'), # Every 10 minutes, offset by 2
id='temperature_check',
misfire_grace_time=30,
max_instances=1
)
# Inventory optimization - every 30 minutes (recommendations)
# Inventory optimization - every 30 minutes (recommendations) - Start at minute 15, 45
self.scheduler.add_job(
self.generate_inventory_recommendations,
CronTrigger(minute='*/30'),
CronTrigger(minute='15,45'), # Offset to avoid conflicts
id='inventory_recs',
misfire_grace_time=120,
max_instances=1
)
# Waste reduction analysis - every hour (recommendations)
# Waste reduction analysis - every hour (recommendations) - Start at minute 30
self.scheduler.add_job(
self.generate_waste_reduction_recommendations,
CronTrigger(minute='0'),
CronTrigger(minute='30'), # Offset to avoid conflicts
id='waste_reduction_recs',
misfire_grace_time=300,
max_instances=1
@@ -96,7 +99,7 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
GREATEST(0, i.low_stock_threshold - COALESCE(SUM(s.current_quantity), 0)) as shortage_amount
FROM ingredients i
LEFT JOIN stock s ON s.ingredient_id = i.id AND s.is_available = true
WHERE i.tenant_id = $1 AND i.is_active = true
WHERE i.tenant_id = :tenant_id AND i.is_active = true
GROUP BY i.id, i.name, i.tenant_id, i.low_stock_threshold, i.max_stock_level, i.reorder_point
)
SELECT * FROM stock_analysis WHERE status != 'normal'
@@ -113,13 +116,16 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
for tenant_id in tenants:
try:
from sqlalchemy import text
async with self.db_manager.get_session() as session:
result = await session.execute(text(query), {"tenant_id": tenant_id})
issues = result.fetchall()
# Add timeout to prevent hanging connections
async with asyncio.timeout(30): # 30 second timeout
async with self.db_manager.get_background_session() as session:
result = await session.execute(text(query), {"tenant_id": tenant_id})
issues = result.fetchall()
for issue in issues:
await self._process_stock_issue(tenant_id, issue)
# Convert SQLAlchemy Row to dictionary for easier access
issue_dict = dict(issue._mapping) if hasattr(issue, '_mapping') else dict(issue)
await self._process_stock_issue(tenant_id, issue_dict)
except Exception as e:
logger.error("Error checking stock for tenant",
@@ -227,18 +233,21 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
ORDER BY s.expiration_date ASC
"""
from sqlalchemy import text
async with self.db_manager.get_session() as session:
result = await session.execute(text(query))
expiring_items = result.fetchall()
# Add timeout to prevent hanging connections
async with asyncio.timeout(30): # 30 second timeout
async with self.db_manager.get_background_session() as session:
result = await session.execute(text(query))
expiring_items = result.fetchall()
# Group by tenant
by_tenant = {}
for item in expiring_items:
tenant_id = item['tenant_id']
# Convert SQLAlchemy Row to dictionary for easier access
item_dict = dict(item._mapping) if hasattr(item, '_mapping') else dict(item)
tenant_id = item_dict['tenant_id']
if tenant_id not in by_tenant:
by_tenant[tenant_id] = []
by_tenant[tenant_id].append(item)
by_tenant[tenant_id].append(item_dict)
for tenant_id, items in by_tenant.items():
await self._process_expiring_items(tenant_id, items)
@@ -328,13 +337,16 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
ORDER BY t.temperature_celsius DESC, t.deviation_minutes DESC
"""
from sqlalchemy import text
async with self.db_manager.get_session() as session:
result = await session.execute(text(query))
breaches = result.fetchall()
# Add timeout to prevent hanging connections
async with asyncio.timeout(30): # 30 second timeout
async with self.db_manager.get_background_session() as session:
result = await session.execute(text(query))
breaches = result.fetchall()
for breach in breaches:
await self._process_temperature_breach(breach)
# Convert SQLAlchemy Row to dictionary for easier access
breach_dict = dict(breach._mapping) if hasattr(breach, '_mapping') else dict(breach)
await self._process_temperature_breach(breach_dict)
except Exception as e:
logger.error("Temperature check failed", error=str(e))
@@ -378,13 +390,13 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
}, item_type='alert')
# Update alert triggered flag to avoid spam
from sqlalchemy import text
async with self.db_manager.get_session() as session:
await session.execute(
text("UPDATE temperature_logs SET alert_triggered = true WHERE id = :id"),
{"id": breach['id']}
)
await session.commit()
# Add timeout to prevent hanging connections
async with asyncio.timeout(10): # 10 second timeout for simple update
async with self.db_manager.get_background_session() as session:
await session.execute(
text("UPDATE temperature_logs SET alert_triggered = true WHERE id = :id"),
{"id": breach['id']}
)
except Exception as e:
logger.error("Error processing temperature breach",
@@ -412,7 +424,7 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
FROM ingredients i
LEFT JOIN stock s ON s.ingredient_id = i.id AND s.is_available = true
LEFT JOIN stock_movements sm ON sm.ingredient_id = i.id
WHERE i.is_active = true AND i.tenant_id = $1
WHERE i.is_active = true AND i.tenant_id = :tenant_id
GROUP BY i.id, i.name, i.tenant_id, i.low_stock_threshold, i.max_stock_level
HAVING COUNT(sm.id) FILTER (WHERE sm.movement_type = 'production_use'
AND sm.created_at > CURRENT_DATE - INTERVAL '30 days') >= 3
@@ -438,12 +450,16 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
for tenant_id in tenants:
try:
from sqlalchemy import text
async with self.db_manager.get_session() as session:
result = await session.execute(text(query), {"tenant_id": tenant_id})
# Add timeout to prevent hanging connections
async with asyncio.timeout(30): # 30 second timeout
async with self.db_manager.get_background_session() as session:
result = await session.execute(text(query), {"tenant_id": tenant_id})
recommendations = result.fetchall()
for rec in recommendations:
await self._generate_stock_recommendation(tenant_id, rec)
# Convert SQLAlchemy Row to dictionary for easier access
rec_dict = dict(rec._mapping) if hasattr(rec, '_mapping') else dict(rec)
await self._generate_stock_recommendation(tenant_id, rec_dict)
except Exception as e:
logger.error("Error generating recommendations for tenant",
@@ -524,7 +540,7 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
JOIN stock_movements sm ON sm.ingredient_id = i.id
WHERE sm.movement_type = 'waste'
AND sm.created_at > CURRENT_DATE - INTERVAL '30 days'
AND i.tenant_id = $1
AND i.tenant_id = :tenant_id
GROUP BY i.id, i.name, i.tenant_id, sm.reason_code
HAVING SUM(sm.quantity) > 5 -- More than 5kg wasted
ORDER BY total_waste_30d DESC
@@ -535,12 +551,16 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
for tenant_id in tenants:
try:
from sqlalchemy import text
async with self.db_manager.get_session() as session:
result = await session.execute(text(query), {"tenant_id": tenant_id})
# Add timeout to prevent hanging connections
async with asyncio.timeout(30): # 30 second timeout
async with self.db_manager.get_background_session() as session:
result = await session.execute(text(query), {"tenant_id": tenant_id})
waste_data = result.fetchall()
for waste in waste_data:
await self._generate_waste_recommendation(tenant_id, waste)
# Convert SQLAlchemy Row to dictionary for easier access
waste_dict = dict(waste._mapping) if hasattr(waste, '_mapping') else dict(waste)
await self._generate_waste_recommendation(tenant_id, waste_dict)
except Exception as e:
logger.error("Error generating waste recommendations",
@@ -703,6 +723,28 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
except Exception as e:
logger.error("Error handling order placed event", error=str(e))
async def get_active_tenants(self) -> List[UUID]:
"""Get list of active tenant IDs from ingredients table (inventory service specific)"""
try:
query = text("SELECT DISTINCT tenant_id FROM ingredients WHERE is_active = true")
# Add timeout to prevent hanging connections
async with asyncio.timeout(10): # 10 second timeout
async with self.db_manager.get_background_session() as session:
result = await session.execute(query)
# Handle PostgreSQL UUID objects properly
tenant_ids = []
for row in result.fetchall():
tenant_id = row.tenant_id
# Convert to UUID if it's not already
if isinstance(tenant_id, UUID):
tenant_ids.append(tenant_id)
else:
tenant_ids.append(UUID(str(tenant_id)))
return tenant_ids
except Exception as e:
logger.error("Error fetching active tenants from ingredients", error=str(e))
return []
async def get_stock_after_order(self, ingredient_id: str, order_quantity: float) -> Optional[Dict[str, Any]]:
"""Get stock information after hypothetical order"""
try:
@@ -710,15 +752,19 @@ class InventoryAlertService(BaseAlertService, AlertServiceMixin):
SELECT i.id, i.name,
COALESCE(SUM(s.current_quantity), 0) as current_stock,
i.low_stock_threshold as minimum_stock,
(COALESCE(SUM(s.current_quantity), 0) - $2) as remaining
(COALESCE(SUM(s.current_quantity), 0) - :order_quantity) as remaining
FROM ingredients i
LEFT JOIN stock s ON s.ingredient_id = i.id AND s.is_available = true
WHERE i.id = $1
WHERE i.id = :ingredient_id
GROUP BY i.id, i.name, i.low_stock_threshold
"""
result = await self.db_manager.fetchrow(query, ingredient_id, order_quantity)
return dict(result) if result else None
# Add timeout to prevent hanging connections
async with asyncio.timeout(10): # 10 second timeout
async with self.db_manager.get_background_session() as session:
result = await session.execute(text(query), {"ingredient_id": ingredient_id, "order_quantity": order_quantity})
row = result.fetchone()
return dict(row) if row else None
except Exception as e:
logger.error("Error getting stock after order",