126 lines
3.8 KiB
Python
126 lines
3.8 KiB
Python
from sqlmodel import Session, select
|
|
from src import messages, models
|
|
from src.products import exceptions
|
|
|
|
|
|
def get_all(
|
|
session: Session,
|
|
user: models.User,
|
|
names: list[str],
|
|
productors: list[str],
|
|
types: list[str],
|
|
) -> list[models.ProductPublic]:
|
|
statement = select(
|
|
models.Product) .join(
|
|
models.Productor,
|
|
models.Product.productor_id == models.Productor.id) .where(
|
|
models.Productor.type.in_(
|
|
[
|
|
r.name for r in user.roles])) .distinct()
|
|
if len(names) > 0:
|
|
statement = statement.where(models.Product.name.in_(names))
|
|
if len(productors) > 0:
|
|
statement = statement.where(models.Productor.name.in_(productors))
|
|
if len(types) > 0:
|
|
statement = statement.where(models.Product.type.in_(types))
|
|
return session.exec(statement.order_by(models.Product.name)).all()
|
|
|
|
|
|
def get_one(
|
|
session: Session,
|
|
product_id: int,
|
|
) -> models.ProductPublic:
|
|
return session.get(models.Product, product_id)
|
|
|
|
|
|
def create_one(
|
|
session: Session,
|
|
product: models.ProductCreate,
|
|
) -> models.ProductPublic:
|
|
if not product:
|
|
raise exceptions.ProductCreateError(
|
|
messages.Messages.invalid_input(
|
|
'product', 'input cannot be None'))
|
|
if not session.get(models.Productor, product.productor_id):
|
|
raise exceptions.ProductorNotFoundError(
|
|
messages.Messages.not_found('productor'))
|
|
product_create = product.model_dump(exclude_unset=True)
|
|
new_product = models.Product(**product_create)
|
|
session.add(new_product)
|
|
session.commit()
|
|
session.refresh(new_product)
|
|
return new_product
|
|
|
|
|
|
def update_one(
|
|
session: Session,
|
|
_id: int,
|
|
product: models.ProductUpdate
|
|
) -> models.ProductPublic:
|
|
statement = select(models.Product).where(models.Product.id == _id)
|
|
result = session.exec(statement)
|
|
new_product = result.first()
|
|
if not new_product:
|
|
raise exceptions.ProductNotFoundError(
|
|
messages.Messages.not_found('product'))
|
|
if product.productor_id and not session.get(
|
|
models.Productor, product.productor_id):
|
|
raise exceptions.ProductorNotFoundError(
|
|
messages.Messages.not_found('productor'))
|
|
|
|
product_updates = product.model_dump(exclude_unset=True)
|
|
for key, value in product_updates.items():
|
|
setattr(new_product, key, value)
|
|
|
|
session.add(new_product)
|
|
session.commit()
|
|
session.refresh(new_product)
|
|
return new_product
|
|
|
|
|
|
def delete_one(
|
|
session: Session,
|
|
_id: int
|
|
) -> models.ProductPublic:
|
|
statement = select(models.Product).where(models.Product.id == _id)
|
|
result = session.exec(statement)
|
|
product = result.first()
|
|
if not product:
|
|
raise exceptions.ProductNotFoundError(
|
|
messages.Messages.not_found('product'))
|
|
result = models.ProductPublic.model_validate(product)
|
|
session.delete(product)
|
|
session.commit()
|
|
return result
|
|
|
|
def is_allowed(
|
|
session: Session,
|
|
user: models.User,
|
|
_id: int = None,
|
|
product: models.ProductCreate = None,
|
|
) -> bool:
|
|
if not _id and not product:
|
|
return False
|
|
if not _id:
|
|
statement = (
|
|
select(models.Product)
|
|
.join(
|
|
models.Productor,
|
|
models.Product.productor_id == models.Productor.id
|
|
)
|
|
.where(models.Product.id == product.productor_id)
|
|
)
|
|
productor = session.exec(statement).first()
|
|
return productor.type in [r.name for r in user.roles]
|
|
statement = (
|
|
select(models.Product)
|
|
.join(
|
|
models.Productor,
|
|
models.Product.productor_id == models.Productor.id
|
|
)
|
|
.where(models.Product.id == _id)
|
|
.where(models.Productor.type.in_([r.name for r in user.roles]))
|
|
.distinct()
|
|
)
|
|
return len(session.exec(statement).all()) > 0
|