This commit is contained in:
Андрей Дувакин 2024-10-04 15:28:46 +05:00
parent 31ea47bf77
commit fa18d94a26
16 changed files with 528 additions and 142 deletions

View File

@ -17,12 +17,13 @@ class AccessoriesService:
AccessoryEntity(
id=a.id,
name=a.name,
width=a.width,
height=a.height,
length=a.length,
volume=a.volume,
weight=a.weight,
period=a.period,
city_id=a.city_id
)
for a in accessories]
for a in accessories
]
def get_accessory_by_id(self, accessory_id: int) -> Optional[AccessoryEntity]:
accessory = self.repository.get_by_id(accessory_id)
@ -30,50 +31,48 @@ class AccessoriesService:
return AccessoryEntity(
id=accessory.id,
name=accessory.name,
width=accessory.width,
height=accessory.height,
length=accessory.length,
volume=accessory.volume,
weight=accessory.weight,
period=accessory.period,
city_id=accessory.city_id
)
return None
def create_accessory(self, entity: AccessoryEntity) -> AccessoryEntity:
accessory_model = Accessory(
name=entity.name,
width=entity.width,
height=entity.height,
length=entity.length,
volume=entity.volume,
weight=entity.weight,
period=entity.period,
city_id=entity.city_id
)
created_accessory = self.repository.create(accessory_model)
return AccessoryEntity(
id=created_accessory.id,
name=created_accessory.name,
width=created_accessory.width,
height=created_accessory.height,
length=created_accessory.length,
volume=created_accessory.volume,
weight=created_accessory.weight,
period=created_accessory.period,
city_id=created_accessory.city_id
)
def update_accessory(self, accessory_id: int, entity: AccessoryEntity) -> Optional[AccessoryEntity]:
accessory_model = self.repository.get_by_id(accessory_id)
if accessory_model:
accessory_model.name = entity.name
accessory_model.width = entity.width
accessory_model.height = entity.height
accessory_model.length = entity.length
accessory_model.volume = entity.volume
accessory_model.weight = entity.weight
accessory_model.period = entity.period
accessory_model.city_id = entity.city_id
self.repository.update(accessory_model)
return AccessoryEntity(
id=accessory_model.id,
name=accessory_model.name,
width=accessory_model.width,
height=accessory_model.height,
length=accessory_model.length,
volume=accessory_model.volume,
weight=accessory_model.weight,
period=accessory_model.period,
city_id=accessory_model.city_id
)
return None
def delete_accessory(self, accessory_id: int) -> bool:

View File

@ -0,0 +1,61 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.core.entities.city import CityEntity
from app.infrastructure.database.models.cities import City
from app.infrastructure.database.repository.city_repository import CitiesRepository
class CitiesService:
def __init__(self, db: Session):
self.repository = CitiesRepository(db)
def get_all_cities(self) -> List[CityEntity]:
cities = self.repository.get_all()
return [
CityEntity(
id=c.id,
name=c.name,
federal_district_id=c.federal_district_id
)
for c in cities
]
def get_city_by_id(self, city_id: int) -> Optional[CityEntity]:
city = self.repository.get_by_id(city_id)
if city:
return CityEntity(
id=city.id,
name=city.name,
federal_district_id=city.federal_district_id
)
return None
def create_city(self, entity: CityEntity) -> CityEntity:
city_model = City(
name=entity.name,
federal_district_id=entity.federal_district_id
)
created_city = self.repository.create(city_model)
return CityEntity(
id=created_city.id,
name=created_city.name,
federal_district_id=created_city.federal_district_id
)
def update_city(self, city_id: int, entity: CityEntity) -> Optional[CityEntity]:
city_model = self.repository.get_by_id(city_id)
if city_model:
city_model.name = entity.name
city_model.federal_district_id = entity.federal_district_id
self.repository.update(city_model)
return CityEntity(
id=city_model.id,
name=city_model.name,
federal_district_id=city_model.federal_district_id
)
return None
def delete_city(self, city_id: int) -> bool:
return self.repository.delete(city_id) is not None

View File

@ -0,0 +1,74 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.core.entities.delivery_accessory import DeliveryAccessoryEntity
from app.infrastructure.database.models.delivery_accessories import DeliveryAccessory
from app.infrastructure.database.repository.delivery_accessory_repository import DeliveryAccessoriesRepository
class DeliveryAccessoriesService:
def __init__(self, db: Session):
self.repository = DeliveryAccessoriesRepository(db)
def get_all_delivery_accessories(self) -> List[DeliveryAccessoryEntity]:
delivery_accessories = self.repository.get_all()
return [
DeliveryAccessoryEntity(
id=da.id,
queue=da.queue,
count=da.count,
accessory_id=da.accessory_id,
delivery_order_id=da.delivery_order_id
)
for da in delivery_accessories
]
def get_delivery_accessory_by_id(self, delivery_accessory_id: int) -> Optional[DeliveryAccessoryEntity]:
delivery_accessory = self.repository.get_by_id(delivery_accessory_id)
if delivery_accessory:
return DeliveryAccessoryEntity(
id=delivery_accessory.id,
queue=delivery_accessory.queue,
count=delivery_accessory.count,
accessory_id=delivery_accessory.accessory_id,
delivery_order_id=delivery_accessory.delivery_order_id
)
return None
def create_delivery_accessory(self, entity: DeliveryAccessoryEntity) -> DeliveryAccessoryEntity:
delivery_accessory_model = DeliveryAccessory(
queue=entity.queue,
count=entity.count,
accessory_id=entity.accessory_id,
delivery_order_id=entity.delivery_order_id
)
created_delivery_accessory = self.repository.create(delivery_accessory_model)
return DeliveryAccessoryEntity(
id=created_delivery_accessory.id,
queue=created_delivery_accessory.queue,
count=created_delivery_accessory.count,
accessory_id=created_delivery_accessory.accessory_id,
delivery_order_id=created_delivery_accessory.delivery_order_id
)
def update_delivery_accessory(self, delivery_accessory_id: int, entity: DeliveryAccessoryEntity) -> Optional[
DeliveryAccessoryEntity]:
delivery_accessory_model = self.repository.get_by_id(delivery_accessory_id)
if delivery_accessory_model:
delivery_accessory_model.queue = entity.queue
delivery_accessory_model.count = entity.count
delivery_accessory_model.accessory_id = entity.accessory_id
delivery_accessory_model.delivery_order_id = entity.delivery_order_id
self.repository.update(delivery_accessory_model)
return DeliveryAccessoryEntity(
id=delivery_accessory_model.id,
queue=delivery_accessory_model.queue,
count=delivery_accessory_model.count,
accessory_id=delivery_accessory_model.accessory_id,
delivery_order_id=delivery_accessory_model.delivery_order_id
)
return None
def delete_delivery_accessory(self, delivery_accessory_id: int) -> bool:
return self.repository.delete(delivery_accessory_id) is not None

View File

@ -0,0 +1,80 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.core.entities.delivery_order import DeliveryOrderEntity
from app.infrastructure.database.models.delivery_orders import DeliveryOrder
from app.infrastructure.database.repository.delivery_order_repository import DeliveryOrdersRepository
class DeliveryOrdersService:
def __init__(self, db: Session):
self.repository = DeliveryOrdersRepository(db)
def get_all_delivery_orders(self) -> List[DeliveryOrderEntity]:
delivery_orders = self.repository.get_all()
return [
DeliveryOrderEntity(
id=o.id,
order_datetime=o.order_datetime,
count_robots=o.count_robots,
deadline=o.deadline,
truck_id=o.truck_id,
total_order_id=o.total_order_id
)
for o in delivery_orders
]
def get_delivery_order_by_id(self, delivery_order_id: int) -> Optional[DeliveryOrderEntity]:
delivery_order = self.repository.get_by_id(delivery_order_id)
if delivery_order:
return DeliveryOrderEntity(
id=delivery_order.id,
order_datetime=delivery_order.order_datetime,
count_robots=delivery_order.count_robots,
deadline=delivery_order.deadline,
truck_id=delivery_order.truck_id,
total_order_id=delivery_order.total_order_id
)
return None
def create_delivery_order(self, entity: DeliveryOrderEntity) -> DeliveryOrderEntity:
delivery_order_model = DeliveryOrder(
order_datetime=entity.order_datetime,
count_robots=entity.count_robots,
deadline=entity.deadline,
truck_id=entity.truck_id,
total_order_id=entity.total_order_id
)
created_delivery_order = self.repository.create(delivery_order_model)
return DeliveryOrderEntity(
id=created_delivery_order.id,
order_datetime=created_delivery_order.order_datetime,
count_robots=created_delivery_order.count_robots,
deadline=created_delivery_order.deadline,
truck_id=created_delivery_order.truck_id,
total_order_id=created_delivery_order.total_order_id
)
def update_delivery_order(self, delivery_order_id: int, entity: DeliveryOrderEntity) -> Optional[
DeliveryOrderEntity]:
delivery_order_model = self.repository.get_by_id(delivery_order_id)
if delivery_order_model:
delivery_order_model.order_datetime = entity.order_datetime
delivery_order_model.count_robots = entity.count_robots
delivery_order_model.deadline = entity.deadline
delivery_order_model.truck_id = entity.truck_id
delivery_order_model.total_order_id = entity.total_order_id
self.repository.update(delivery_order_model)
return DeliveryOrderEntity(
id=delivery_order_model.id,
order_datetime=delivery_order_model.order_datetime,
count_robots=delivery_order_model.count_robots,
deadline=delivery_order_model.deadline,
truck_id=delivery_order_model.truck_id,
total_order_id=delivery_order_model.total_order_id
)
return None
def delete_delivery_order(self, delivery_order_id: int) -> bool:
return self.repository.delete(delivery_order_id) is not None

View File

@ -0,0 +1,59 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.core.entities.federal_district import FederalDistrictEntity
from app.infrastructure.database.models.federal_districts import FederalDistrict
from app.infrastructure.database.repository.federal_district_repository import FederalDistrictsRepository
class FederalDistrictsService:
def __init__(self, db: Session):
self.repository = FederalDistrictsRepository(db)
def get_all_federal_districts(self) -> List[FederalDistrictEntity]:
federal_districts = self.repository.get_all()
return [
FederalDistrictEntity(
id=fd.id,
name=fd.name,
cities=[city.id for city in fd.cities] if fd.cities else []
)
for fd in federal_districts
]
def get_federal_district_by_id(self, federal_district_id: int) -> Optional[FederalDistrictEntity]:
federal_district = self.repository.get_by_id(federal_district_id)
if federal_district:
return FederalDistrictEntity(
id=federal_district.id,
name=federal_district.name,
cities=[city.id for city in federal_district.cities] if federal_district.cities else []
)
return None
def create_federal_district(self, entity: FederalDistrictEntity) -> FederalDistrictEntity:
federal_district_model = FederalDistrict(
name=entity.name
)
created_federal_district = self.repository.create(federal_district_model)
return FederalDistrictEntity(
id=created_federal_district.id,
name=created_federal_district.name,
cities=[] # Поскольку города еще не добавлены
)
def update_federal_district(self, federal_district_id: int, entity: FederalDistrictEntity) -> Optional[FederalDistrictEntity]:
federal_district_model = self.repository.get_by_id(federal_district_id)
if federal_district_model:
federal_district_model.name = entity.name
self.repository.update(federal_district_model)
return FederalDistrictEntity(
id=federal_district_model.id,
name=federal_district_model.name,
cities=[city.id for city in federal_district_model.cities] if federal_district_model.cities else []
)
return None
def delete_federal_district(self, federal_district_id: int) -> bool:
return self.repository.delete(federal_district_id) is not None

View File

@ -1,69 +0,0 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.core.entities.order import OrderEntity
from app.infrastructure.database.models.total_orders import TotalOrder
from app.infrastructure.database.repository.order_repository import OrderRepository
class OrderService:
def __init__(self, db: Session):
self.repository = OrderRepository(db)
def get_all_orders(self) -> List[OrderEntity]:
orders = self.repository.get_all()
return [
OrderEntity(
id=o.id,
order_datetime=o.order_datetime,
user_id=o.user_id,
status_id=o.status_id,
)
for o in orders
]
def get_order_by_id(self, order_id: int) -> Optional[OrderEntity]:
order = self.repository.get_by_id(order_id)
if order:
return OrderEntity(
id=order.id,
order_datetime=order.order_datetime,
user_id=order.user_id,
status_id=order.status_id,
)
return None
def create_order(self, entity: OrderEntity) -> OrderEntity:
order_model = TotalOrder(
order_datetime=entity.order_datetime,
user_id=entity.user_id,
status_id=entity.status_id,
)
created_order = self.repository.create(order_model)
return OrderEntity(
id=created_order.id,
order_datetime=created_order.order_datetime,
user_id=created_order.user_id,
status_id=created_order.status_id,
)
def update_order(self, order_id: int, entity: OrderEntity) -> Optional[OrderEntity]:
order_model = self.repository.get_by_id(order_id)
if order_model:
order_model.order_datetime = entity.order_datetime
order_model.user_id = entity.user_id
order_model.status_id = entity.status_id
self.repository.update(order_model)
return OrderEntity(
id=order_model.id,
order_datetime=order_model.order_datetime,
user_id=order_model.user_id,
status_id=order_model.status_id,
)
return None
def delete_order(self, order_id: int) -> bool:
return self.repository.delete(order_id) is not None

View File

@ -4,20 +4,22 @@ from sqlalchemy.orm import Session
from app.core.entities.role import RoleEntity
from app.infrastructure.database.models.roles import Role
from app.infrastructure.database.repository.role_repository import RoleRepository
from app.infrastructure.database.repository.role_repository import RolesRepository
class RoleService:
class RolesService:
def __init__(self, db: Session):
self.repository = RoleRepository(db)
self.repository = RolesRepository(db)
def get_all_roles(self) -> List[RoleEntity]:
roles = self.repository.get_all()
return [
RoleEntity(
id=r.id,
name=r.name,
) for r in roles
id=role.id,
name=role.name,
users=[user.id for user in role.users] if role.users else []
)
for role in roles
]
def get_role_by_id(self, role_id: int) -> Optional[RoleEntity]:
@ -26,16 +28,19 @@ class RoleService:
return RoleEntity(
id=role.id,
name=role.name,
users=[user.id for user in role.users] if role.users else []
)
return None
def create_role(self, entity: RoleEntity) -> RoleEntity:
role_model = Role(name=entity.name)
role_model = Role(
name=entity.name
)
created_role = self.repository.create(role_model)
return RoleEntity(
id=created_role.id,
name=created_role.name,
users=[]
)
def update_role(self, role_id: int, entity: RoleEntity) -> Optional[RoleEntity]:
@ -46,8 +51,8 @@ class RoleService:
return RoleEntity(
id=role_model.id,
name=role_model.name,
users=[user.id for user in role_model.users] if role_model.users else []
)
return None
def delete_role(self, role_id: int) -> bool:

View File

@ -4,21 +4,23 @@ from sqlalchemy.orm import Session
from app.core.entities.status import StatusEntity
from app.infrastructure.database.models.statuses import Status
from app.infrastructure.database.repository.status_repository import StatusRepository
from app.infrastructure.database.repository.status_repository import StatusesRepository
class StatusService:
class StatusesService:
def __init__(self, db: Session):
self.repository = StatusRepository(db)
self.repository = StatusesRepository(db)
def get_all_statuses(self) -> List[StatusEntity]:
statuses = self.repository.get_all()
return [
StatusEntity(
id=s.id,
name=s.name,
id=status.id,
name=status.name,
orders=[order.id for order in status.orders] if status.orders else [],
steps=[step.id for step in status.steps] if status.steps else []
)
for s in statuses
for status in statuses
]
def get_status_by_id(self, status_id: int) -> Optional[StatusEntity]:
@ -27,16 +29,21 @@ class StatusService:
return StatusEntity(
id=status.id,
name=status.name,
orders=[order.id for order in status.orders] if status.orders else [],
steps=[step.id for step in status.steps] if status.steps else []
)
return None
def create_status(self, entity: StatusEntity) -> StatusEntity:
status_model = Status(name=entity.name)
status_model = Status(
name=entity.name
)
created_status = self.repository.create(status_model)
return StatusEntity(
id=created_status.id,
name=created_status.name,
orders=[],
steps=[]
)
def update_status(self, status_id: int, entity: StatusEntity) -> Optional[StatusEntity]:
@ -47,8 +54,9 @@ class StatusService:
return StatusEntity(
id=status_model.id,
name=status_model.name,
orders=[order.id for order in status_model.orders] if status_model.orders else [],
steps=[step.id for step in status_model.steps] if status_model.steps else []
)
return None
def delete_status(self, status_id: int) -> bool:

View File

@ -0,0 +1,83 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.core.entities.total_order import TotalOrderEntity
from app.infrastructure.database.models.total_orders import TotalOrder
from app.infrastructure.database.repository.total_order_repository import TotalOrdersRepository
class TotalOrdersService:
def __init__(self, db: Session):
self.repository = TotalOrdersRepository(db)
def get_all_total_orders(self) -> List[TotalOrderEntity]:
total_orders = self.repository.get_all()
return [
TotalOrderEntity(
id=order.id,
order_datetime=order.order_datetime,
count_robots=order.count_robots,
deadline=order.deadline,
user_id=order.user_id,
status_id=order.status_id,
delivery_orders=[do.id for do in order.delivery_orders] if order.delivery_orders else []
)
for order in total_orders
]
def get_total_order_by_id(self, total_order_id: int) -> Optional[TotalOrderEntity]:
total_order = self.repository.get_by_id(total_order_id)
if total_order:
return TotalOrderEntity(
id=total_order.id,
order_datetime=total_order.order_datetime,
count_robots=total_order.count_robots,
deadline=total_order.deadline,
user_id=total_order.user_id,
status_id=total_order.status_id,
delivery_orders=[do.id for do in total_order.delivery_orders] if total_order.delivery_orders else []
)
return None
def create_total_order(self, entity: TotalOrderEntity) -> TotalOrderEntity:
total_order_model = TotalOrder(
order_datetime=entity.order_datetime,
count_robots=entity.count_robots,
deadline=entity.deadline,
user_id=entity.user_id,
status_id=entity.status_id
)
created_total_order = self.repository.create(total_order_model)
return TotalOrderEntity(
id=created_total_order.id,
order_datetime=created_total_order.order_datetime,
count_robots=created_total_order.count_robots,
deadline=created_total_order.deadline,
user_id=created_total_order.user_id,
status_id=created_total_order.status_id,
delivery_orders=[]
)
def update_total_order(self, total_order_id: int, entity: TotalOrderEntity) -> Optional[TotalOrderEntity]:
total_order_model = self.repository.get_by_id(total_order_id)
if total_order_model:
total_order_model.order_datetime = entity.order_datetime
total_order_model.count_robots = entity.count_robots
total_order_model.deadline = entity.deadline
total_order_model.user_id = entity.user_id
total_order_model.status_id = entity.status_id
self.repository.update(total_order_model)
return TotalOrderEntity(
id=total_order_model.id,
order_datetime=total_order_model.order_datetime,
count_robots=total_order_model.count_robots,
deadline=total_order_model.deadline,
user_id=total_order_model.user_id,
status_id=total_order_model.status_id,
delivery_orders=[do.id for do in total_order_model.delivery_orders] if total_order_model.delivery_orders else []
)
return None
def delete_total_order(self, total_order_id: int) -> bool:
return self.repository.delete(total_order_id) is not None

View File

@ -0,0 +1,69 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.core.entities.truck import TruckEntity
from app.infrastructure.database.models.trucks import Truck
from app.infrastructure.database.repository.truck_repository import TrucksRepository
class TrucksService:
def __init__(self, db: Session):
self.repository = TrucksRepository(db)
def get_all_trucks(self) -> List[TruckEntity]:
trucks = self.repository.get_all()
return [
TruckEntity(
id=truck.id,
name=truck.name,
capacity=truck.capacity,
volume=truck.volume,
delivery_orders=[do.id for do in truck.delivery_orders] if truck.delivery_orders else []
)
for truck in trucks
]
def get_truck_by_id(self, truck_id: int) -> Optional[TruckEntity]:
truck = self.repository.get_by_id(truck_id)
if truck:
return TruckEntity(
id=truck.id,
name=truck.name,
capacity=truck.capacity,
volume=truck.volume,
delivery_orders=[do.id for do in truck.delivery_orders] if truck.delivery_orders else []
)
return None
def create_truck(self, entity: TruckEntity) -> TruckEntity:
truck_model = Truck(
name=entity.name,
capacity=entity.capacity,
volume=entity.volume
)
created_truck = self.repository.create(truck_model)
return TruckEntity(
id=created_truck.id,
name=created_truck.name,
capacity=created_truck.capacity,
volume=created_truck.volume,
delivery_orders=[]
)
def update_truck(self, truck_id: int, entity: TruckEntity) -> Optional[TruckEntity]:
truck_model = self.repository.get_by_id(truck_id)
if truck_model:
truck_model.name = entity.name
truck_model.capacity = entity.capacity
truck_model.volume = entity.volume
self.repository.update(truck_model)
return TruckEntity(
id=truck_model.id,
name=truck_model.name,
capacity=truck_model.capacity,
volume=truck_model.volume,
delivery_orders=[do.id for do in truck_model.delivery_orders] if truck_model.delivery_orders else []
)
return None
def delete_truck(self, truck_id: int) -> bool:
return self.repository.delete(truck_id) is not None

View File

@ -1,45 +1,31 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.core.entities.user import UserEntity
from app.infrastructure.database.models.users import User
from app.infrastructure.database.repository.user_repository import UserRepository
from app.infrastructure.database.repository.user_repository import UsersRepository
class UserService:
class UsersService:
def __init__(self, db: Session):
self.repository = UserRepository(db)
self.repository = UsersRepository(db)
def get_all_users(self) -> List[UserEntity]:
users = self.repository.get_all()
return [
UserEntity(
user.id,
user.first_name,
user.last_name,
user.login,
user.password,
user.role_id
) for user in users
id=user.id,
first_name=user.first_name,
last_name=user.last_name,
login=user.login,
password=user.password,
role_id=user.role_id,
total_orders=[to.id for to in user.total_orders] if user.total_orders else []
)
for user in users
]
def get_user_by_id(self, user_id: int) -> Optional[UserEntity]:
user = self.repository.get_by_id(user_id)
if user:
return UserEntity(
user.id,
user.first_name,
user.last_name,
user.login,
user.password,
user.role_id
)
return None
def get_user_by_login(self, login: str) -> Optional[UserEntity]:
user = self.repository.get_by_login(login)
if user:
return UserEntity(
id=user.id,
@ -47,9 +33,9 @@ class UserService:
last_name=user.last_name,
login=user.login,
password=user.password,
role_id=user.role_id
role_id=user.role_id,
total_orders=[to.id for to in user.total_orders] if user.total_orders else []
)
return None
def create_user(self, entity: UserEntity) -> UserEntity:
@ -58,7 +44,7 @@ class UserService:
last_name=entity.last_name,
login=entity.login,
password=entity.password,
role_id=entity.role_id,
role_id=entity.role_id
)
created_user = self.repository.create(user_model)
return UserEntity(
@ -68,6 +54,7 @@ class UserService:
login=created_user.login,
password=created_user.password,
role_id=created_user.role_id,
total_orders=[]
)
def update_user(self, user_id: int, entity: UserEntity) -> Optional[UserEntity]:
@ -85,9 +72,9 @@ class UserService:
last_name=user_model.last_name,
login=user_model.login,
password=user_model.password,
role_id=user_model.role_id
role_id=user_model.role_id,
total_orders=[to.id for to in user_model.total_orders] if user_model.total_orders else []
)
return None
def delete_user(self, user_id: int) -> bool:

View File

@ -23,6 +23,6 @@ def init_db():
import app.infrastructure.database.models.accessory
import app.infrastructure.database.models.total_orders
import app.infrastructure.database.models.delivery_orders
import app.infrastructure.database.models.dilevery_accessiories
import app.infrastructure.database.models.delivery_accessories
Base.metadata.create_all(bind=engine)

View File

@ -1,8 +1,8 @@
from sqlalchemy.orm import Session
from app.infrastructure.database.models.dilevery_accessiories import DeliveryAccessory
from app.infrastructure.database.models.delivery_accessories import DeliveryAccessory
class DeliveryAccessoryRepository:
class DeliveryAccessoriesRepository:
def __init__(self, db: Session):
self.db = db

View File

@ -0,0 +1,30 @@
from sqlalchemy.orm import Session
from app.infrastructure.database.models.federal_districts import FederalDistrict
class FederalDistrictsRepository:
def __init__(self, db: Session):
self.db = db
def get_all(self):
return self.db.query(FederalDistrict).all()
def get_by_id(self, federal_district_id: int):
return self.db.query(FederalDistrict).filter(FederalDistrict.id == federal_district_id).first()
def create(self, federal_district: FederalDistrict):
self.db.add(federal_district)
self.db.commit()
self.db.refresh(federal_district)
return federal_district
def update(self, federal_district: FederalDistrict):
self.db.commit()
return federal_district
def delete(self, federal_district_id: int):
federal_district = self.db.query(FederalDistrict).filter(FederalDistrict.id == federal_district_id).first()
if federal_district:
self.db.delete(federal_district)
self.db.commit()
return federal_district
return None