121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
from sqlalchemy.orm import Session
|
|
from models import Car, EngineType, HybridType, PowerRatio, Personal
|
|
import schemas
|
|
from typing import List, Optional
|
|
|
|
def get_car(db: Session, car_id: int) -> Optional[Car]:
|
|
return db.query(Car).filter(Car.id == car_id).first()
|
|
|
|
def get_cars(db: Session, skip: int = 0, limit: int = 100) -> List[Car]:
|
|
return db.query(Car).offset(skip).limit(limit).all()
|
|
|
|
def get_cars_count(db: Session) -> int:
|
|
return db.query(Car).count()
|
|
|
|
def create_car(db: Session, car: schemas.CarCreate) -> Car:
|
|
# Преобразование из Pydantic enum в SQLAlchemy enum
|
|
engine_type = EngineType(car.engine_type.value)
|
|
hybrid_type = HybridType(car.hybrid_type.value)
|
|
power_ratio = PowerRatio(car.power_ratio.value)
|
|
|
|
db_car = Car(
|
|
image=car.image,
|
|
name=car.name,
|
|
price=car.price,
|
|
base_price=car.base_price,
|
|
country_of_origin=car.country_of_origin,
|
|
year=car.year,
|
|
drive_type=car.drive_type,
|
|
mileage=car.mileage,
|
|
engine_capacity=car.engine_capacity,
|
|
engine_power=car.engine_power,
|
|
engine_type=engine_type,
|
|
electric_motor_power=car.electric_motor_power,
|
|
hybrid_type=hybrid_type,
|
|
power_ratio=power_ratio,
|
|
)
|
|
db.add(db_car)
|
|
db.commit()
|
|
db.refresh(db_car)
|
|
return db_car
|
|
|
|
def update_car(db: Session, car_id: int, car_update: schemas.CarUpdate) -> Optional[Car]:
|
|
db_car = get_car(db, car_id)
|
|
if not db_car:
|
|
return None
|
|
|
|
update_data = car_update.model_dump(exclude_unset=True)
|
|
|
|
# Обработка enum полей
|
|
if "engine_type" in update_data and update_data["engine_type"] is not None:
|
|
update_data["engine_type"] = EngineType(update_data["engine_type"])
|
|
|
|
if "hybrid_type" in update_data and update_data["hybrid_type"] is not None:
|
|
update_data["hybrid_type"] = HybridType(update_data["hybrid_type"])
|
|
|
|
if "power_ratio" in update_data and update_data["power_ratio"] is not None:
|
|
update_data["power_ratio"] = PowerRatio(update_data["power_ratio"])
|
|
|
|
# Обновление полей
|
|
for key, value in update_data.items():
|
|
setattr(db_car, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_car)
|
|
return db_car
|
|
|
|
def delete_car(db: Session, car_id: int) -> bool:
|
|
db_car = get_car(db, car_id)
|
|
if not db_car:
|
|
return False
|
|
|
|
db.delete(db_car)
|
|
db.commit()
|
|
return True
|
|
|
|
# Персонал ---------
|
|
|
|
def get_personal(db: Session, personal_id: int) -> Optional[Personal]:
|
|
return db.query(Personal).filter(Personal.id == personal_id).first()
|
|
|
|
def get_all_personal(db: Session, skip: int = 0, limit: int = 100) -> List[Personal]:
|
|
return db.query(Personal).offset(skip).limit(limit).all()
|
|
|
|
def get_personal_count(db: Session) -> int:
|
|
return db.query(Personal).count()
|
|
|
|
def create_personal(db: Session, personal: schemas.PersonalCreate) -> Personal:
|
|
db_personal = Personal(
|
|
name=personal.name,
|
|
surname=personal.surname,
|
|
role=personal.role,
|
|
photo=personal.photo
|
|
)
|
|
db.add(db_personal)
|
|
db.commit()
|
|
db.refresh(db_personal)
|
|
return db_personal
|
|
|
|
def update_personal(db: Session, personal_id: int, personal_update: schemas.PersonalUpdate) -> Optional[Personal]:
|
|
db_personal = get_personal(db, personal_id)
|
|
if not db_personal:
|
|
return None
|
|
|
|
update_data = personal_update.model_dump(exclude_unset=True)
|
|
|
|
# Обновление полей
|
|
for key, value in update_data.items():
|
|
setattr(db_personal, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_personal)
|
|
return db_personal
|
|
|
|
def delete_personal(db: Session, personal_id: int) -> bool:
|
|
db_personal = get_personal(db, personal_id)
|
|
if not db_personal:
|
|
return False
|
|
|
|
db.delete(db_personal)
|
|
db.commit()
|
|
return True |