Chris-May

    def get(self, reference) -> model.Batch:
        [[batch_id, *data]] = self.session.execute(
            'SELECT * FROM batches WHERE reference=:reference',
            dict(reference=reference)
        )
        batch = model.Batch(*data)
        allocations = self.session.execute(
            'SELECT o.orderid, o.sku, o.qty'
            ' FROM allocations al INNER JOIN order_lines o'
            ' ON al.orderline_id = o.id'

Chris-May

Prerequisites for JS compilation

• npm
• webpack
• webpack-bundle-tracker
• django-webpack-loader

brew install npm
npm install --save-dev webpack webpack-cli webpack-bundle-tracker \

Chris-May

from uuid import uuid4

from eventsourcing.application.command import CommandProcess
from eventsourcing.application.system import System
from eventsourcing.domain.model.aggregate import AggregateRoot
from eventsourcing.domain.model.entity import DomainEntity
from eventsourcing.application.process import Process
from eventsourcing.domain.model.command import Command
from eventsourcing.domain.model.decorators import attribute, retry
from eventsourcing.exceptions import OperationalError, RecordConflictError

Chris-May

from uuid import uuid4

from eventsourcing.application.simple import SimpleApplication
from eventsourcing.domain.model.aggregate import AggregateRoot
from eventsourcing.domain.model.decorators import attribute, subscribe_to
from eventsourcing.domain.model.entity import TimestampedVersionedEntity, DomainEntity
from eventsourcing.domain.model.events import publish


app = SimpleApplication(uri='sqlite:///db.sqlite')

Chris-May

from datetime import datetime
from uuid import uuid4

from eventsourcing.application.simple import SimpleApplication
from eventsourcing.domain.model.aggregate import AggregateRoot
from eventsourcing.domain.model.decorators import attribute
from eventsourcing.domain.model.entity import TimestampedVersionedEntity
from eventsourcing.domain.model.events import publish

app = SimpleApplication(uri='sqlite:///db.sqlite')