HW 06

.flake8

@@ -1,2 +1,4 @@
 [flake8]
 extend-ignore = E203,E501
+exclude =
+    tests

.github/workflows/main.yaml

@@ -91,4 +91,33 @@ jobs:
       - name: Run pytest
         run: |
           cd homework_05
+          poetry run pytest tests
+
+  hw_06:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: actions/setup-python@v2
+        with:
+          python-version: 3.12
+
+      - name: Run image # install poetry
+        uses: abatilo/actions-poetry@v2.0.0
+        with:
+          poetry-version: 1.8.2
+
+      - name: Install dependencies # install all dependencies
+        run: |
+          cd homework_06
+          poetry install --no-root
+
+      - name: Run static analyzis
+        run: |
+          cd homework_06
+          poetry run pre-commit run --files *
+
+      - name: Run pytest
+        run: |
+          cd homework_06
           poetry run pytest tests

.pre-commit-config.yaml

@@ -38,6 +38,7 @@ repos:
     -   id: mypy
         exclude: 'tests'
 
+
 # run local pylint in venv
 -   repo: local
     hooks:
@@ -51,7 +52,7 @@ repos:
         - --max-line-length=120
         - --ignore-imports=yes
         - -d duplicate-code
-        - -d C0111,W0621,R0913,R1705,W0201,W0613
+        - -d C0111,W0621,R0913,R1705,W0201,R0903,W0613
 
 
 -   repo: https://github.com/asottile/pyupgrade

homework_06/domain/models.py

@@ -0,0 +1,19 @@
+from dataclasses import dataclass, field
+from typing import List
+
+
+@dataclass
+class Product:
+    id: int
+    name: str
+    quantity: int
+    price: float
+
+
+@dataclass
+class Order:
+    id: int
+    products: List[Product] = field(default_factory=list)
+
+    def add_product(self, product: Product):
+        self.products.append(product)

homework_06/domain/repositories.py

@@ -0,0 +1,32 @@
+from abc import ABC, abstractmethod
+from typing import List
+
+from .models import Order, Product
+
+
+class ProductRepository(ABC):
+    @abstractmethod
+    def add(self, product: Product):
+        pass
+
+    @abstractmethod
+    def get(self, product_id: int) -> Product:
+        pass
+
+    @abstractmethod
+    def list(self) -> List[Product]:
+        pass
+
+
+class OrderRepository(ABC):
+    @abstractmethod
+    def add(self, order: Order):
+        pass
+
+    @abstractmethod
+    def get(self, order_id: int) -> Order:
+        pass
+
+    @abstractmethod
+    def list(self) -> List[Order]:
+        pass

homework_06/domain/services.py

@@ -0,0 +1,27 @@
+from typing import List
+
+from .models import Order, Product
+from .repositories import OrderRepository, ProductRepository
+
+
+def generate_next_id():
+    num = 1
+    while True:
+        yield num
+        num += 1
+
+
+class WarehouseService:
+    def __init__(self, product_repo: ProductRepository, order_repo: OrderRepository):
+        self.product_repo = product_repo
+        self.order_repo = order_repo
+
+    def create_product(self, name: str, quantity: int, price: float) -> Product:
+        product = Product(id=generate_next_id(), name=name, quantity=quantity, price=price)
+        self.product_repo.add(product)
+        return product
+
+    def create_order(self, products: List[Product]) -> Order:
+        order = Order(id=generate_next_id(), products=products)
+        self.order_repo.add(order)
+        return order

homework_06/domain/unit_of_work.py

@@ -0,0 +1,19 @@
+from abc import ABC, abstractmethod
+
+
+class UnitOfWork(ABC):
+    @abstractmethod
+    def __enter__(self):
+        pass
+
+    @abstractmethod
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        pass
+
+    @abstractmethod
+    def commit(self):
+        pass
+
+    @abstractmethod
+    def rollback(self):
+        pass

homework_06/infrastructure/database.py

@@ -0,0 +1 @@
+DATABASE_URL = "sqlite:///warehouse.db"

homework_06/infrastructure/orm.py

@@ -0,0 +1,28 @@
+# mypy: ignore-errors
+from sqlalchemy import Column, Float, ForeignKey, Integer, String, Table
+from sqlalchemy.orm import declarative_base, relationship
+
+Base = declarative_base()
+
+
+class ProductORM(Base):
+    __tablename__ = "products"
+    id = Column(Integer, primary_key=True)
+    name = Column(String)
+    quantity = Column(Integer)
+    price = Column(Float)
+
+
+class OrderORM(Base):
+    __tablename__ = "orders"
+    id = Column(Integer, primary_key=True)
+
+
+order_product_associations = Table(
+    "order_product_associations",
+    Base.metadata,
+    Column("order_id", ForeignKey("orders.id")),
+    Column("product_id", ForeignKey("products.id")),
+)
+
+OrderORM.products = relationship("ProductORM", secondary=order_product_associations)

homework_06/infrastructure/repositories.py

@@ -0,0 +1,61 @@
+from typing import List
+
+from sqlalchemy.orm import Session
+
+from domain.models import Order, Product
+from domain.repositories import OrderRepository, ProductRepository
+
+from .orm import OrderORM, ProductORM
+
+
+class SqlAlchemyProductRepository(ProductRepository):
+    def __init__(self, session: Session):
+        self.session = session
+
+    def add(self, product: Product):
+        product_orm = ProductORM(name=product.name, quantity=product.quantity, price=product.price)
+        self.session.add(product_orm)
+
+    def get(self, product_id: int) -> Product:
+        product_orm = self.session.query(ProductORM).filter_by(id=product_id).one()
+        return Product(
+            id=int(product_orm.id),
+            name=str(product_orm.name),
+            quantity=int(product_orm.quantity),
+            price=float(product_orm.price),
+        )
+
+    def list(self) -> List[Product]:
+        products_orm = self.session.query(ProductORM).all()
+        return [
+            Product(
+                id=int(p.id),
+                name=str(p.name),
+                quantity=int(p.quantity),
+                price=float(p.price),
+            )
+            for p in products_orm
+        ]
+
+
+class SqlAlchemyOrderRepository(OrderRepository):
+    def __init__(self, session: Session):
+        self.session = session
+
+    def add(self, order: Order):
+        order_orm = OrderORM()
+        order_orm.products = [self.session.query(ProductORM).filter_by(id=p.id).one() for p in order.products]
+        self.session.add(order_orm)
+
+    def get(self, order_id: int) -> Order:
+        order_orm = self.session.query(OrderORM).filter_by(id=order_id).one()
+        products = [Product(id=p.id, name=p.name, quantity=p.quantity, price=p.price) for p in order_orm.products]
+        return Order(id=int(order_orm.id), products=list(products))
+
+    def list(self) -> List[Order]:
+        orders_orm = self.session.query(OrderORM).all()
+        orders = []
+        for order_orm in orders_orm:
+            products = [Product(id=p.id, name=p.name, quantity=p.quantity, price=p.price) for p in order_orm.products]
+            orders.append(Order(id=int(order_orm.id), products=products))
+        return orders

homework_06/infrastructure/unit_of_work.py

@@ -0,0 +1,23 @@
+from domain.unit_of_work import UnitOfWork
+from sqlalchemy.orm import Session
+
+
+class SqlAlchemyUnitOfWork(UnitOfWork):
+    def __init__(self, session: Session):
+        self.session = session
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if exc_type is None:
+            self.commit()
+        else:
+            self.rollback()
+        self.session.close()
+
+    def commit(self):
+        self.session.commit()
+
+    def rollback(self):
+        self.session.rollback()

homework_06/main.py

@@ -0,0 +1,40 @@
+# isort: skip_file
+from domain.services import WarehouseService
+from infrastructure.database import DATABASE_URL
+from infrastructure.orm import Base
+from infrastructure.repositories import (
+    SqlAlchemyOrderRepository,
+    SqlAlchemyProductRepository,
+)
+from infrastructure.unit_of_work import SqlAlchemyUnitOfWork
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+engine = create_engine(DATABASE_URL)
+SessionFactory = sessionmaker(bind=engine)
+Base.metadata.create_all(engine)
+
+
+def main():
+    session = SessionFactory()
+    product_repo = SqlAlchemyProductRepository(session)
+    order_repo = SqlAlchemyOrderRepository(session)
+
+    uow = SqlAlchemyUnitOfWork(session)
+
+    warehouse_service = WarehouseService(product_repo, order_repo)
+    with uow:
+        new_product = warehouse_service.create_product(name="test1", quantity=1, price=100)
+        uow.commit()
+        print(f"create product: {new_product}")
+
+        another_product = warehouse_service.create_product(name="test2", quantity=10, price=42)
+        uow.commit()
+        print(f"create product: {another_product}")
+
+        # decided to revert
+        uow.rollback()
+
+
+if __name__ == "__main__":
+    main()