/
woodpecker86
/
cosmic-python-code
Обзор
Документация
Войти
/
woodpecker86
/
cosmic-python-code
Код
Запросы
0
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/allocation/adapters/repository.py
59 строк
2 KB
Harry
Use the ORM instead [view_using_orm]
01 мар 2024, 18:05
01 мар 2024, 18:05
d0593d3
Код
Авторство
О чём код?
import abc from typing import Set from allocation.adapters import orm from allocation.domain import model class AbstractRepository(abc.ABC): def __init__(self): self.seen = set() # type: Set[model.Product] def add(self, product: model.Product): self._add(product) self.seen.add(product) def get(self, sku) -> model.Product: product = self._get(sku) if product: self.seen.add(product) return product def get_by_batchref(self, batchref) -> model.Product: product = self._get_by_batchref(batchref) if product: self.seen.add(product) return product @abc.abstractmethod def _add(self, product: model.Product): raise NotImplementedError @abc.abstractmethod def _get(self, sku) -> model.Product: raise NotImplementedError @abc.abstractmethod def _get_by_batchref(self, batchref) -> model.Product: raise NotImplementedError class SqlAlchemyRepository(AbstractRepository): def __init__(self, session): super().__init__() self.session = session def _add(self, product): self.session.add(product) def _get(self, sku): return self.session.query(model.Product).filter_by(sku=sku).first() def _get_by_batchref(self, batchref): return ( self.session.query(model.Product) .join(model.Batch) .filter( orm.batches.c.reference == batchref, ) .first() )