58 lines
1.9 KiB
Python
58 lines
1.9 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from app.models.models import Object, SLAContract, Task, Incident, Customer
|
|
|
|
|
|
async def get_object_by_name_or_id(db: AsyncSession, query: str) -> Object:
|
|
try:
|
|
obj_id = int(query)
|
|
return await db.get(Object, obj_id)
|
|
except (ValueError, TypeError):
|
|
result = await db.execute(
|
|
select(Object).where(Object.name.ilike(f"%{query}%")).limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_sla_for_object(db: AsyncSession, object_id: int) -> SLAContract:
|
|
result = await db.execute(
|
|
select(SLAContract).where(
|
|
SLAContract.object_id == object_id,
|
|
SLAContract.status == "active"
|
|
).order_by(SLAContract.created_at.desc()).limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_objects_for_customer(db: AsyncSession, customer_name: str) -> list:
|
|
result = await db.execute(
|
|
select(Customer).where(Customer.name.ilike(f"%{customer_name}%")).limit(1)
|
|
)
|
|
customer = result.scalar_one_or_none()
|
|
if not customer:
|
|
return []
|
|
result = await db.execute(
|
|
select(Object).where(Object.customer_id == customer.id, Object.status == "active")
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def get_active_tasks_for_object(db: AsyncSession, object_id: int) -> list:
|
|
result = await db.execute(
|
|
select(Task).where(
|
|
Task.object_id == object_id,
|
|
Task.status.in_(["open", "in_progress"])
|
|
)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def get_last_incident_for_object(db: AsyncSession, object_id: int):
|
|
result = await db.execute(
|
|
select(Incident).where(
|
|
Incident.object_id == object_id,
|
|
Incident.status.in_(["resolved", "closed"])
|
|
).order_by(Incident.created_at.desc()).limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|