webapp/app/services/auto_booking_service.py

from sqlalchemy.orm import Session
from sqlalchemy import func
from app.models.auto_booking import AutoBooking
from app.schemas.auto_booking import AutoBookingCreate
from typing import List

class AutoBookingService:

    @staticmethod
    def create(db: Session, obj_in: AutoBookingCreate) -> AutoBooking:
        db_obj = AutoBooking(**obj_in.dict())
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    @staticmethod
    def get_by_id(db: Session, id: int):
        return db.query(AutoBooking).filter(AutoBooking.id == id).first()

    @staticmethod
    def delete_by_id(db: Session, id: int):
        obj = db.query(AutoBooking).filter(AutoBooking.id == id).first()
        if obj:
            db.delete(obj)
            db.commit()
            return True
        return False

    @staticmethod
    def update_by_id(db: Session, id: int, updated_data: dict):
        obj = db.query(AutoBooking).filter(AutoBooking.id == id).first()
        if not obj:
            return None
        for key, value in updated_data.items():
            setattr(obj, key, value)
        db.commit()
        db.refresh(obj)
        return obj

    @staticmethod
    def get_paginated(db: Session, tech_provider: str, keyword: str, page: int, size: int):
        query = db.query(AutoBooking)
        if tech_provider:
            query = query.filter(AutoBooking.provider == tech_provider)
        if keyword:
            like_str = f"%{keyword}%"
            query = query.filter(
                (AutoBooking.first_name.like(like_str)) |
                (AutoBooking.last_name.like(like_str)) |
                (AutoBooking.email.like(like_str)) |
                (AutoBooking.visa_center.like(like_str))
            )
        return query.offset(page * size).limit(size).all()

    @staticmethod
    def batch_get_by_ids(db: Session, ids: List[int]):
        return db.query(AutoBooking).filter(AutoBooking.id.in_(ids)).all()

    @staticmethod
    def statistics(db: Session, tech_provider: str):
        query = db.query(AutoBooking.provider, func.count(AutoBooking.id)).group_by(AutoBooking.provider)
        if tech_provider:
            query = query.filter(AutoBooking.provider == tech_provider)
        return query.all()

    @staticmethod
    def get_pending(db: Session, tech_provider: str):
        query = db.query(AutoBooking).filter(AutoBooking.status == 0)
        if tech_provider:
            query = query.filter(AutoBooking.provider == tech_provider)
        return query.all()