POC with dataclasses

.gitignore

@@ -1,3 +1,4 @@
 __pycache__
 *.egg-info
 tmp/
+db/

kaba/__init__.py

@@ -1,4 +1,3 @@
-import os
 from time import perf_counter
 
 import ujson as json
@@ -6,11 +5,10 @@ import hupper
 import minicli
 from bson import ObjectId
 from jinja2 import Environment, PackageLoader, select_autoescape
-from pymongo import MongoClient
 from roll import Roll, Response
 from roll.extensions import cors, options, traceback, simple_server
 
-from .base import Document, Str, Float, Array, Email, Int, Reference, Datetime, Mapping
+from .models import Delivery, Order, Person, Product, ProductOrder
 
 
 class Response(Response):
@@ -45,65 +43,6 @@ env = Environment(
 )
 
 
-class Producer(Document):
-    __collection__ = "producers"
-    name = Str(required=True)
-
-    @property
-    def products(self):
-        return Product.find(producer=self._id)
-
-
-class Product(Document):
-    __collection__ = "products"
-    producer = Reference(Producer, required=True)
-    name = Str(required=True)
-    ref = Str(required=True)
-    description = Str()
-    price = Float(required=True)
-
-
-class Person(Document):
-    __collection__ = "persons"
-    first_name = Str()
-    last_name = Str()
-    email = Email()
-
-
-class ProductOrder(Document):
-    wanted = Int()
-    ordered = Int()
-
-
-class PersonOrder(Document):
-    person = Str()
-    products = Mapping(str, ProductOrder)
-
-    def get_quantity(self, product):
-        choice = self.products.get(product.ref)
-        return choice.wanted if choice else 0
-
-    def total(self, products):
-        products = {p.ref: p for p in products}
-        return round(
-            sum(p.wanted * products[ref].price for ref, p in self.products.items()), 2
-        )
-
-
-class Order(Document):
-    __collection__ = "orders"
-    when = Datetime(required=True)
-    where = Str()
-    producer = Reference(Producer, required=True)
-    products = Array(Product)
-    orders = Mapping(str, PersonOrder)
-
-    def product_wanted(self, product):
-        return round(
-            sum([po.products[product.ref].wanted for po in self.orders.values()])
-        )
-
-
 app = Roll()
 cors(app, methods="*", headers="*")
 options(app)
@@ -121,66 +60,88 @@ async def on_startup():
 
 @app.route("/", methods=["GET"])
 async def home(request, response):
-    response.html("home.html", {"orders": Order.find()})
+    response.html("home.html", deliveries=Delivery.all())
 
 
-@app.route("/commande/{order_id}/total", methods=["GET"])
+@app.route("/livraison/new", methods=["GET"])
-async def view_order(request, response, order_id):
+async def new_delivery(request, response):
-    order = Order.find_one(_id=ObjectId(order_id))
+    response.html("edit_delivery.html", delivery={})
-    total = round(sum(po.total(order.products) for po in order.orders.values()), 2)
-    response.html(
-        "order.html",
-        {
-            "order": order,
-            "producer": Producer.find_one(_id=order.producer),
-            "total": total,
-        },
-    )
 
 
-@app.route("/commande/{order_id}", methods=["GET"])
+@app.route("/livraison/new", methods=["POST"])
-async def order_form(request, response, order_id):
+async def create_delivery(request, response):
-    order = Order.find_one(_id=ObjectId(order_id))
-    email = request.query.get("email")
-    person_order = order.orders.get(email)
-    response.html(
-        "place_order.html",
-        {
-            "order": order,
-            "person": email,
-            "person_order": person_order,
-            "producer": Producer.find_one(_id=order.producer),
-        },
-    )
-
-
-@app.route("/commande/{order_id}", methods=["POST"])
-async def place_order(request, response, order_id):
-    order = Order.find_one(_id=ObjectId(order_id))
-    email = request.query.get("email")
-    person_order = PersonOrder(person=email)
     form = request.form
-    for product in order.products:
+    data = {}
+    for name, field in Delivery.__dataclass_fields__.items():
+        if name in form:
+            data[name] = form.get(name)
+    delivery = Delivery(**data)
+    delivery.persist()
+    response.status = 302
+    response.headers["Location"] = f"/livraison/{delivery.id}"
+
+
+@app.route("/livraison/{id}/edit", methods=["GET"])
+async def edit_delivery(request, response, id):
+    delivery = Delivery.load(id)
+    response.html("edit_delivery.html", {"delivery": delivery})
+
+
+@app.route("/livraison/{id}/edit", methods=["POST"])
+async def post_delivery(request, response, id):
+    delivery = Delivery.load(id)
+    form = request.form
+    for name, field in Delivery.__dataclass_fields__.items():
+        if name in form:
+            setattr(delivery, name, form.get(name))
+    delivery.persist()
+    response.status = 302
+    response.headers["Location"] = f"/livraison/{delivery.id}"
+
+
+@app.route("/livraison/{id}", methods=["GET"])
+async def view_delivery(request, response, id):
+    delivery = Delivery.load(id)
+    total = round(sum(o.total(delivery.products) for o in delivery.orders.values()), 2)
+    response.html("delivery.html", {"delivery": delivery, "total": total})
+
+
+@app.route("/livraison/{id}/commander", methods=["GET"])
+async def order_form(request, response, id):
+    delivery = Delivery.load(id)
+    email = request.query.get("email")
+    order = delivery.orders.get(email)
+    response.html(
+        "place_order.html", {"delivery": delivery, "person": email, "order": order}
+    )
+
+
+@app.route("/livraison/{id}/commander", methods=["POST"])
+async def place_order(request, response, id):
+    delivery = Delivery.load(id)
+    email = request.query.get("email")
+    order = Order()
+    form = request.form
+    for product in delivery.products:
         quantity = form.int(product.ref, 0)
         if quantity:
-            person_order.products[product.ref] = ProductOrder(wanted=quantity)
+            order.products[product.ref] = ProductOrder(wanted=quantity)
-    if not order.orders:
+    if not delivery.orders:
-        order.orders = {}
+        delivery.orders = {}
-    order.orders[email] = person_order
+    delivery.orders[email] = order
-    order.replace_one()
+    delivery.persist()
     response.headers["Location"] = request.url.decode()
     response.status = 302
 
 
 def connect():
-    db = os.environ.get("KABA_DB", "mongodb://localhost/kaba")
+    # db = os.environ.get("KABA_DB", "mongodb://localhost/kaba")
-    client = MongoClient(db)
+    # client = MongoClient(db)
-    db = client.get_database()
+    # db = client.get_database()
-    Producer.bind(db)
+    # Person.bind(db)
-    Product.bind(db)
+    # Delivery.bind(db)
-    Order.bind(db)
+    # return client
-    Person.bind(db)
+    pass
-    return client
 
 
 @minicli.cli()
@@ -194,11 +155,11 @@ def shell():
         start_ipython(
             argv=[],
             user_ns={
-                "Producer": Producer,
                 "app": app,
                 "Product": Product,
                 "Person": Person,
                 "Order": Order,
+                "Delivery": Delivery,
             },
         )
 

kaba/base.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import datetime, date
 
 from bson import ObjectId
 
@@ -63,6 +63,17 @@ class Datetime(Field):
             return datetime.fromtimestamp(value)
 
 
+class Date(Field):
+    @staticmethod
+    def coerce(value):
+        if isinstance(value, date):
+            return value
+        if isinstance(value, datetime):
+            return value.date()
+        if isinstance(value, int):
+            return date.fromtimestamp(value)
+
+
 class Email(Field):
     @staticmethod
     def coerce(value):
@@ -126,10 +137,13 @@ class Array(Field):
 
 class MetaDocument(type):
     def __new__(cls, name, bases, attrs):
+        fields = getattr(cls, "_fields", {})
         for attr_name, attr_value in attrs.items():
             if not isinstance(attr_value, Field):
                 continue
+            fields[attr_name] = attr_value
             attr_value.name = attr_name
+        cls._fields = fields
         return super().__new__(cls, name, bases, attrs)
 
 
@@ -137,9 +151,6 @@ class Document(dict, metaclass=MetaDocument):
     __db__ = None
     __collection__ = None
 
-    # def __repr__(self):
-    #     return f"<{self.__class__.__name__} {self._id}>"
-
     def __init__(self, data=None, **attrs):
         if data:
             for key, value in data.items():

kaba/templates/home.html

@@ -1,15 +1,15 @@
 {% extends "base.html" %}
 
 {% block body %}
-    {% for order in orders %}
+    {% for delivery in deliveries %}
-    <p>Producteur: {{ order.producer }}</p>
+    <p>Producteur: {{ delivery.producer }}</p>
-    <p>Lieu: {{ order.where }}</p>
+    <p>Lieu: {{ delivery.where }}</p>
-    <p>Date: {{ order.when }}</p>
+    <p>Date: {{ delivery.when }}</p>
-    <form action="/commande/{{ order._id }}">
+    <form action="/livraison/{{ delivery.id }}/commander">
-        <label for="email">Participer à la commande</label>
+        <label for="email">Commander</label>
-        <input type="text" name="email">
+        <input type="email" name="email">
         <input type="submit" value="Commander">
     </form>
-    <a href="/commande/{{ order._id }}/total">Résumé de la commande</a>
+    <a href="/livraison/{{ delivery.id }}">Résumé de la livraison</a>
     {% endfor %}
 {% endblock body %}

kaba/templates/order.html

@@ -1,33 +0,0 @@
-{% extends "base.html" %}
-
-{% block body %}
-<p><a href="/">&lt; Commandes</a></p>
-<p>Producteur: {{ producer.name }}</p>
-<p>Lieu: {{ order.where }}</p>
-<p>Date: {{ order.when }}</p>
-<table>
-    <tr>
-        <th>Produit</th>
-        <th>Prix</th>
-        {% for email, person_order in order.orders.items() %}
-            <th><a href="/commande/{{ order._id }}?email={{ email }}">{{ email }}</a></th>
-        {% endfor %}
-        <th>Total</th>
-    </tr>
-    {% for product in order.products %}
-    <tr>
-        <th>{{ product.name }}</th>
-        <td>{{ product.price }} €</td>
-        {% for email, person_order in order.orders.items() %}
-            {% if product.ref in person_order.products %}
-            <td>{{ person_order.products[product.ref].wanted }}</td>
-            {% else %}
-                <td>—</td>
-            {% endif %}
-        {% endfor %}
-        <th>{{ order.product_wanted(product) }}</th>
-    </tr>
-    {% endfor %}
-</table>
-<p>Total: {{ total }} €</p>
-{% endblock body %}

kaba/templates/place_order.html

@@ -2,18 +2,18 @@
 
 {% block body %}
 <p><a href="/">&lt; Commandes</a></p>
-<p>Producteur: {{ producer.name }}</p>
+<p>Producteur: {{ delivery.producer }}</p>
-<p>Lieu: {{ order.where }}</p>
+<p>Lieu: {{ delivery.where }}</p>
-<p>Date: {{ order.when }}</p>
+<p>Date: {{ delivery.when }}</p>
 <p>Commande de «{{ person }}»</p>
 <form method="post">
     <table>
         <tr><th>Produit</th><th>Prix</th><th>Quantité</th></tr>
-        {% for product in order.products %}
+        {% for product in delivery.products %}
-        <tr><th>{{ product.name }}</th><td>{{ product.price }} €</td><td><input type="number" name="{{ product.ref }}" value="{{ person_order.get_quantity(product) if person_order else 0 }}"></td></tr>
+        <tr><th>{{ product.name }}</th><td>{{ product.price }} €</td><td><input type="number" name="{{ product.ref }}" value="{{ order.get_quantity(product) if order else 0 }}"></td></tr>
         {% endfor %}
     </table>
-    <p>Total: {{ person_order.total(order.products) if person_order else 0 }} €</p>
+    <p>Total: {{ order.total(delivery.products) if order else 0 }} €</p>
     <input type="hidden" name="email" value="{{ person }}">
     <input type="submit" value="Valider ma commande">
 </form>

tests/conftest.py

@@ -3,26 +3,36 @@ import os
 import pytest
 from roll.extensions import traceback
 
-# Before loading any eurordis module.
-os.environ["KABA_DB"] = "mongodb://localhost/kaba_test"
-
 from kaba import app as kaba_app
-from kaba import connect, Producer
+from kaba import utils
+from kaba.models import Delivery, Person
 
 
 def pytest_configure(config):
-    client = connect()
+    from kaba import models
-    client.drop_database("test_kaba")
+    models.Base.ROOT = "tmp/db"
 
 
-def pytest_runtest_setup(item):
+# def pytest_runtest_setup(item):
-    # assert get_db().name == "test_eurordis"
+#     # assert get_db().name == "test_eurordis"
-    for cls in [Producer]:
+#     for cls in [Delivery]:
-        collection = cls.collection
+#         collection = cls.collection
-        collection.drop()
+#         collection.drop()
 
 
 @pytest.fixture
 def app():  # Requested by Roll testing utilities.
     traceback(kaba_app)
     return kaba_app
+
+
+@pytest.fixture
+def delivery():
+    return Delivery(
+        producer="Andines", when=utils.utcnow(), order_before=utils.utcnow()
+    )
+
+
+@pytest.fixture
+def person():
+    return Person(email="foo@bar.fr")

tests/test_models.py

@@ -1,23 +1,72 @@
-from kaba import Producer, Order, Product
+import pytest
+
+from kaba import utils
+from kaba.models import Delivery, Product, Person, Order, ProductOrder
 
 
-def test_can_create_producer():
+def test_can_create_delivery():
-    producer = Producer(name="Andines")
+    delivery = Delivery(
-    producer.insert_one()
+        producer="Andines", when=utils.utcnow(), order_before=utils.utcnow()
-    assert producer.name == "Andines"
+    )
-    retrieved = Producer.find_one(name="Andines")
+    assert delivery.producer == "Andines"
-    assert retrieved.name == producer.name
+    assert delivery.where == "Marché de la Briche"
-    assert retrieved._id == producer._id
+    assert delivery.when.year == utils.utcnow().year
+    assert delivery.id
 
 
-def test_can_create_order():
+def test_wrong_datetime_raise_valueerror():
-    order = Order(products=[Product(name="riz", price="2.4")])
+    with pytest.raises(ValueError):
-    order.insert_one()
+        Delivery(producer="Andines", order_before=utils.utcnow(), when="pouet")
-    retrieved = Order.find_one(_id=order._id)
-    assert retrieved.products[0].name == "riz"
 
 
-def test_can_update_order_products():
+def test_can_create_product():
-    order = Order()
+    product = Product(name="Lait 1.5L", ref="123", price=1.5)
-    order.products.append(Product(name="riz", price="2.4"))
+    assert product.ref == "123"
+    assert product.price == 1.5
+
+
+def test_can_create_delivery_with_products():
+    delivery = Delivery(
+        producer="Andines",
+        when=utils.utcnow(),
+        order_before=utils.utcnow(),
+        products=[Product(name="Lait", ref="123", price=1.5)],
+    )
+    assert len(delivery.products) == 1
+    assert delivery.products[0].ref == "123"
+
+
+def test_can_add_product_to_delivery(delivery):
+    assert not delivery.products
+    delivery.products.append(Product(name="Chocolat", ref="choco", price=10))
+    assert len(delivery.products) == 1
+
+
+def test_can_create_person():
+    person = Person(email="foo@bar.fr", first_name="Foo")
+    assert person.email == "foo@bar.fr"
+    assert person.first_name == "Foo"
+
+
+def test_can_create_order_with_products():
+    order = Order(products={"123": ProductOrder(wanted=2)})
     assert len(order.products) == 1
+    assert order.products["123"].wanted == 2
+
+
+def test_can_add_product_to_order():
+    order = Order()
+    assert len(order.products) == 0
+    order.products["123"] = ProductOrder(wanted=2)
+    assert order.products["123"].wanted == 2
+
+
+def test_can_persist_delivery(delivery):
+    delivery.persist()
+
+
+def test_can_load_delivery(delivery):
+    delivery.producer = "Corto"
+    delivery.persist()
+    loaded = Delivery.load(delivery.id)
+    assert loaded.producer == "Corto"