### Installation and Setup (Bash) Source: https://context7.com/cosmicpython/cosmicpython.github.io/llms.txt Install Python dependencies using pip and generate the static site. Then, start a local development server to view the site. ```bash # Install dependencies pip install -r requirements.txt # Requirements: markdown, jinja2, pygments, lxml[cssselect] # Generate blog posts and index ./generate-html.py # Start local server and view at http://localhost:8899 python -m http.server 8899 ``` -------------------------------- ### Example Usage of Fake Repository Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_02_repository.html Demonstrates how to instantiate and use a fake repository in tests. This setup allows for isolated testing of application logic. ```python fake_repo = FakeRepository([batch1, batch2, batch3]) ``` -------------------------------- ### Project Directory Structure Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_04_service_layer.html This is an example of a project directory structure that separates concerns into distinct folders like domain, service_layer, adapters, and entrypoints. ```python . ├── config.py ├── domain #(1) │ ├── __init__.py │ └── model.py ├── service_layer #(2) │ ├── __init__.py │ └── services.py ├── adapters #(3) │ ├── __init__.py │ ├── orm.py │ └── repository.py ├── entrypoints (4) │ ├── __init__.py │ └── flask_app.py └── tests ├── __init__.py ├── conftest.py ├── unit │ ├── test_allocate.py │ ├── test_batches.py │ └── test_services.py ├── integration │ ├── test_orm.py │ └── test_repository.py └── e2e └── test_api.py ``` -------------------------------- ### Python Package Setup Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/appendix_project_structure.html Configures a Python project to be installable via pip. Specify the package name and version, and list the subfolders to be included as top-level modules. ```python from setuptools import setup setup( name="allocation", version="0.1", packages=["allocation"], ) ``` -------------------------------- ### Python Package Setup Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html Configures a Python project to be installable via pip. Specify the package name and the subfolders to be included as top-level modules. ```python from setuptools import setup setup( name='allocation', version='0.1', packages=['allocation'], ) ``` -------------------------------- ### Repository Pattern Example Source: https://context7.com/cosmicpython/cosmicpython.github.io/llms.txt Demonstrates a basic implementation of the Repository pattern for database interactions. ```Python class FooRepository: def __init__(self, db_session): self.session = db_session def add_new_item(self, item): self.db_session.add(item) def get_item(self, id): return self.db_session.get(Foo, id) ``` -------------------------------- ### Configuration Registry Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/posts/2019-08-03-ioc-techniques.md Illustrates the Configuration Registry pattern where 'hello_world' retrieves its 'output_function' from a global 'config' dictionary. This allows external configuration. ```python # hello_world.py config = {} def hello_world(): output_function = config["OUTPUT_FUNCTION"] output_function("Hello, world.") ``` -------------------------------- ### Calling Hello World Function Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/blog/2019-08-03-ioc-techniques.html Demonstrates how to call the basic 'hello_world' function from a main script. This setup lacks dependency injection. ```python # main.py from hello_world import hello_world if __name__ == "__main__": hello_world() ``` -------------------------------- ### Python Project Directory Structure Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html This is an example of a project directory structure for a Python application, separating concerns into domain, service_layer, adapters, entrypoints, and tests. ```python . ├── config.py ├── domain (1) │ ├── __init__.py │ └── model.py ├── service_layer (2) │ ├── __init__.py │ └── services.py ├── adapters (3) │ ├── __init__.py │ ├── orm.py │ └── repository.py ├── entrypoints (4) │ ├── __init__.py │ └── flask_app.py └── tests ├── __init__.py ├── conftest.py ├── unit │ ├── test_allocate.py │ ├── test_batches.py │ └── test_services.py ├── integration │   ├── test_orm.py │   └── test_repository.py └── e2e    └── test_api.py ``` -------------------------------- ### Initialize DI in Integration Tests Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_13_dependency_injection.html Use `bootstrap.bootstrap()` with overridden defaults for a custom message bus in integration tests. This example shows overriding defaults for starting the ORM, using a specific Unit of Work, and faking out email and publish functionalities. ```python import pytest from allocation import bootstrap from allocation.adapters import unit_of_work from allocation.domain import commands from allocation.service_layer import views from sqlalchemy.orm import clear_mappers @pytest.fixture def sqlite_bus(sqlite_session_factory): bus = bootstrap.bootstrap( start_orm=True, #(1) uow=unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory), send_mail=lambda *args: None, #(3) publish=lambda *args: None, #(3) ) yield bus clear_mappers() def test_allocations_view(sqlite_bus): sqlite_bus.handle(commands.CreateBatch("sku1batch", "sku1", 50, None)) sqlite_bus.handle(commands.CreateBatch("sku2batch", "sku2", 50, today)) ... assert views.allocations("order1", sqlite_bus.uow) == [ {"sku": "sku1", "batchref": "sku1batch"}, {"sku": "sku2", "batchref": "sku2batch"}, ] ``` -------------------------------- ### Basic FooRepository Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/posts/2017-09-08-repository-and-unit-of-work-pattern-in-python.md A placeholder for a basic repository pattern implementation. This snippet is intended to be expanded upon for specific data access logic. ```python class FooRepository: ``` -------------------------------- ### Python Application Dockerfile Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/appendix_project_structure.html This Dockerfile sets up a Python 3.9 environment, installs dependencies, copies source code, and configures a default Flask application startup command. Install dependencies in order of least to most frequently changing to maximize Docker build cache. ```docker FROM python:3.9-slim-buster # RUN apt install gcc libpq (no longer needed bc we use psycopg2-binary) COPY requirements.txt /tmp/ RUN pip install -r /tmp/requirements.txt RUN mkdir -p /src COPY src/ /src/ RUN pip install -e /src COPY tests/ /tests/ WORKDIR /src ENV FLASK_APP=allocation/entrypoints/flask_app.py FLASK_DEBUG=1 PYTHONUNBUFFERED=1 CMD flask run --host=0.0.0.0 --port=80 ``` -------------------------------- ### Service Layer Test Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html This example demonstrates rewriting the domain-level test to operate through the service layer, interacting with a repository and session. ```python def test_prefers_warehouse_batches_to_shipments(): in_stock_batch = Batch("in-stock-batch", "RETRO-CLOCK", 100, eta=None) shipment_batch = Batch("shipment-batch", "RETRO-CLOCK", 100, eta=tomorrow) repo = FakeRepository([in_stock_batch, shipment_batch]) session = FakeSession() line = OrderLine('oref', "RETRO-CLOCK", 10) services.allocate(line, repo, session) assert in_stock_batch.available_quantity == 90 assert shipment_batch.available_quantity == 100 ``` -------------------------------- ### Function-based Command Handler Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/blog/2017-09-07-introducing-command-handler.html A simple function can also serve as a command handler. This example demonstrates adding a new issue to a log. ```python def ReportIssue(issue_log, cmd): reported_by = IssueReporter( cmd.reporter_name, cmd.reporter_email) issue = Issue(reported_by, cmd.problem_description) issue_log.add(issue) ``` -------------------------------- ### Initialize and Use Message Bus Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/posts/2017-09-19-why-use-domain-events.md Instantiate a message bus and subscribe a command to its handler. This is a basic setup for message dispatching. ```python bus = MessageBus() bus.subscribe_to(ReportIssueCommand, ReportIssueHandler(db.unit_of_work_manager)) bus.handle(cmd) ``` -------------------------------- ### Subscriber Registry Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/posts/2019-08-03-ioc-techniques.md A 'hello_people' function that iterates over a global 'people' list and prints a greeting for each person. This demonstrates the Subscriber Registry pattern where the list can be populated by multiple sources. ```python # hello_people.py people = [] def hello_people(): for person in people: print(f"Hello, {person}.") ``` -------------------------------- ### Clone and Checkout Code Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html Clone the repository and checkout the specific branch for the chapter to follow along with the code examples. ```bash git clone https://github.com/cosmicpython/code.git cd code git checkout chapter_04_service_layer ``` ```bash # or to code along, checkout Chapter 2: git checkout chapter_02_repository ``` -------------------------------- ### Initialize DI in Unit Tests Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_13_dependency_injection.html For unit tests, reuse `FakeUnitOfWork` and disable ORM startup. This example fakes out email and Redis adapters by providing no-op functions. ```python from allocation import bootstrap from allocation.adapters.fake_unit_of_work import FakeUnitOfWork def bootstrap_test_app(): return bootstrap.bootstrap( start_orm=False, #(1) uow=FakeUnitOfWork(), #(2) send_mail=lambda *args: None, #(3) publish=lambda *args: None, #(3) ) ``` -------------------------------- ### Clone Chapter 6 Code from GitHub Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_06_uow.html Clone the repository and checkout the specific branch for chapter 6 to access the code examples. ```bash git clone https://github.com/cosmicpython/code.git cd code git checkout chapter_06_uow ``` -------------------------------- ### Domain Service Function Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html Illustrates a domain service function that does not require object instantiation. Useful for stateless operations or when a full object-oriented approach is not necessary. ```python def add(a: int, b: int) -> int: return a + b ``` -------------------------------- ### Concrete Port Implementations Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/blog/2017-09-08-repository-and-unit-of-work-pattern-in-python.html Provides example implementations of abstract ReadablePort and WriteablePort interfaces. These classes represent different types of devices that can be plugged into a system, such as a light detector or a buzzer. ```Python class LightDetector(ReadablePort): def read(self): return self.get_light_amplitude() class Buzzer(WriteablePort): def write(self, value): if value > 0: self.make_infuriating_noise() class Dial(ReadablePort): def read(self): return self.current_value class Light(self): def write(self, value): if value > 0: self.on = True else: self.on = False ``` -------------------------------- ### Initialize DI in Integration Tests Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html Use `bootstrap.bootstrap()` with overridden defaults for a custom message bus in integration tests. This example overrides the default UoW, send_mail, and publish functions. ```python import pytest from allocation import bootstrap from allocation.adapters import unit_of_work from allocation.domain import commands from allocation.service_layer import views from sqlalchemy.orm import clear_mappers @pytest.fixture def sqlite_bus(sqlite_session_factory): bus = bootstrap.bootstrap( start_orm=True, (1) uow=unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory), send_mail=lambda *args: None, (3) publish=lambda *args: None, (3) ) yield bus clear_mappers() def test_allocations_view(sqlite_bus): sqlite_bus.handle(commands.CreateBatch('sku1batch', 'sku1', 50, None)) sqlite_bus.handle(commands.CreateBatch('sku2batch', 'sku2', 50, today)) ... assert views.allocations('order1', sqlite_bus.uow) == [ {'sku': 'sku1', 'batchref': 'sku1batch'}, {'sku': 'sku2', 'batchref': 'sku2batch'}, ] ``` -------------------------------- ### SQLAlchemy Declarative Syntax Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_02_repository.html This SQLAlchemy example shows a model definition that directly inherits from ORM classes, leading to model dependency on the ORM. Ensure SQLAlchemy is installed to use this syntax. ```python from sqlalchemy import Column, ForeignKey, Integer, String from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import relationship Base = declarative_base() class Order(Base): id = Column(Integer, primary_key=True) class OrderLine(Base): id = Column(Integer, primary_key=True) sku = Column(String(250)) qty = Integer(String(250)) order_id = Column(Integer, ForeignKey('order.id')) order = relationship(Order) class Allocation(Base): ... ``` -------------------------------- ### Complete System Listing: Domain, Service, and Adapters Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/blog/2017-09-07-introducing-command-handler.html This comprehensive Python code demonstrates the domain model, service layer with a command handler, and an adapter implementation using a fake issue log. It includes a test case for reporting an issue. ```python from typing import NamedTuple from expects import expect, have_len, equal # Domain model class IssueReporter: def __init__(self, name, email): self.name = name self.email = email class Issue: def __init__(self, reporter, description): self.description = description self.reporter = reporter class IssueLog: def add(self, issue): pass class ReportIssueCommand(NamedTuple): reporter_name: str reporter_email: str problem_description: str # Service Layer class ReportIssueHandler: def __init__(self, issue_log): self.issue_log = issue_log def __call__(self, cmd): reported_by = IssueReporter( cmd.reporter_name, cmd.reporter_email) issue = Issue(reported_by, cmd.problem_description) self.issue_log.add(issue) # Adapters class FakeIssueLog(IssueLog): def __init__(self): self.issues = [] def add(self, issue): self.issues.append(issue) def get(self, id): return self.issues[id] def __len__(self): return len(self.issues) def __getitem__(self, idx): return self.issues[idx] email = "bob@example.org" name = "bob" desc = "My mouse won't move" class When_reporting_an_issue: def given_an_empty_issue_log(self): self.issues = FakeIssueLog() def because_we_report_a_new_issue(self): handler = ReportIssueHandler(self.issues) cmd = ReportIssueCommand(name, email, desc) handler(cmd) def the_handler_should_have_created_a_new_issue(self): expect(self.issues).to(have_len(1)) def it_should_have_recorded_the_issuer(self): expect(self.issues[0].reporter.name).to(equal(name)) expect(self.issues[0].reporter.email).to(equal(email)) def it_should_have_recorded_the_description(self): expect(self.issues[0].description).to(equal(desc)) ``` -------------------------------- ### Makefile Build Commands Source: https://context7.com/cosmicpython/cosmicpython.github.io/llms.txt Makefile provides convenient commands for building the site, updating book content, and running a local development server. Use 'make help' for a list of targets. ```makefile # Build blog posts and index page make build # Runs: ./generate-html.py # Start local development server on port 8899 make serve # Runs: python -m http.server 8899 # Watch for file changes and rebuild automatically make watch-build # Runs: ls **/*.md **/*.html **/*.xml *.py | entr ./generate-html.py # Update book content from source repository (assumes ../book exists) make update-book # Runs: # cd ../book && make html # ./copy-and-fix-book-html.py # rsync -a -v ../book/images/ ./book/images/ # Complete build with book update and server make all # Runs: build update-book serve ``` -------------------------------- ### Domain Layer Test Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_05_high_gear_low_gear.html An example of a unit test written directly against the domain model's internal state. ```python # domain-layer test: def test_prefers_current_stock_batches_to_shipments(): in_stock_batch = Batch("in-stock-batch", "RETRO-CLOCK", 100, eta=None) shipment_batch = Batch("shipment-batch", "RETRO-CLOCK", 100, eta=tomorrow) line = OrderLine("oref", "RETRO-CLOCK", 10) allocate(line, [in_stock_batch, shipment_batch]) assert in_stock_batch.available_quantity == 90 assert shipment_batch.available_quantity == 100 ``` -------------------------------- ### Configure Hello World Output Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/blog/2019-08-03-ioc-techniques.html Uses a dictionary as a configuration registry to set the output function for 'hello_world'. The registry is populated externally. ```python # hello_world.py config = {} def hello_world(): output_function = config["OUTPUT_FUNCTION"] output_function("Hello, world.") ``` ```python # main.py import hello_world hello_world.config["OUTPUT_FUNCTION"] = print if __name__ == "__main__": hello_world.hello_world() ``` -------------------------------- ### Domain Model Test Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html This is an example of a unit test directly targeting the domain model's behavior for batch allocation. ```python def test_prefers_current_stock_batches_to_shipments(): in_stock_batch = Batch("in-stock-batch", "RETRO-CLOCK", 100, eta=None) shipment_batch = Batch("shipment-batch", "RETRO-CLOCK", 100, eta=tomorrow) line = OrderLine("oref", "RETRO-CLOCK", 10) allocate(line, [in_stock_batch, shipment_batch]) assert in_stock_batch.available_quantity == 90 assert shipment_batch.available_quantity == 100 ``` -------------------------------- ### Main Script with Dependency Injection Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/posts/2019-08-03-ioc-techniques.md The main script configured to inject the 'print' function into 'hello_world'. This demonstrates how to wire up dependencies externally. ```python # main.py import hello_world if __name__ == "__main__": hello_world.hello_world(output_function=print) ``` -------------------------------- ### Soft Delete Implementation Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html Instead of direct deletion, a soft-delete pattern is often used, typically by marking records as cancelled or inactive. This example shows a conceptual `cancel` method. ```python class Batch: def __init__(self, batch_id, items): self.batchid = batchid self.items = items self.cancelled = False def cancel(self): self.cancelled = True ``` -------------------------------- ### Basic Hello World Function Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/blog/2019-08-03-ioc-techniques.html A simple Python function that prints 'Hello, world.' to the console. This serves as a base for demonstrating dependency injection. ```python # hello_world.py def hello_world(): print("Hello, world.") ``` -------------------------------- ### Service Layer Function to Add Batch (Implicit Commit Example) Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_06_uow.html Example of a service layer function that benefits from an implicitly committing Unit of Work, reducing the need for explicit commit calls. ```python def add_batch(ref: str, sku: str, qty: int, eta: Optional[date], uow): with uow: uow.batches.add(model.Batch(ref, sku, qty, eta)) # uow.commit() ``` -------------------------------- ### Subscribe to Hello World Event Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/blog/2019-08-03-ioc-techniques.html Uses a list of callables as a subscriber registry to react to an event. Subscribers are executed after the main event. ```python # hello_world.py subscribers = [] def hello_world(): print("Hello, world.") for subscriber in subscribers: subscriber() ``` ```python # log.py import hello_world def write_to_log(): ... hello_world.subscribers.append(write_to_log) ``` -------------------------------- ### API Test: Happy Path Allocation Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html Tests a successful allocation scenario by POSTing an order and then GETting the allocation status. Asserts a 202 Accepted status for the POST and verifies the allocation details in the GET response. ```python @pytest.mark.usefixtures('postgres_db') @pytest.mark.usefixtures('restart_api') def test_happy_path_returns_202_and_batch_is_allocated(): orderid = random_orderid() sku, othersku = random_sku(), random_sku('other') earlybatch = random_batchref(1) laterbatch = random_batchref(2) otherbatch = random_batchref(3) api_client.post_to_add_batch(laterbatch, sku, 100, '2011-01-02') api_client.post_to_add_batch(earlybatch, sku, 100, '2011-01-01') api_client.post_to_add_batch(otherbatch, othersku, 100, None) r = api_client.post_to_allocate(orderid, sku, qty=3) assert r.status_code == 202 r = api_client.get_allocation(orderid) assert r.ok assert r.json() == [ {'sku': sku, 'batchref': earlybatch}, ] ``` -------------------------------- ### Flask Global Variable Scope Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html This example demonstrates a global variable within a Flask application's module scope. Be cautious, as this can cause issues with in-process testing using Flask's Test Client. ```python # flask_app.py (conceptual) from flask import Flask app = Flask(__name__) # Global variable in module scope my_global_var = "initial_value" @app.route('/') def hello(): global my_global_var return f"Hello! Global var is: {my_global_var}" if __name__ == '__main__': # Consider using Flask app factories for better testability # app.run() pass ``` -------------------------------- ### Project Directory Structure Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/appendix_project_structure.html Illustrates the typical file and directory layout for the project, including Docker, Makefiles, and source code organization. ```tree . ├── Dockerfile (1) ├── Makefile (2) ├── README.md ├── docker-compose.yml (1) ├── license.txt ├── mypy.ini ├── requirements.txt ├── src (3) │ ├── allocation │ │ ├── __init__.py │ │ ├── adapters │ │ │ ├── __init__.py │ │ │ ├── orm.py │ │ │ └── repository.py │ │ ├── config.py │ │ ├── domain │ │ │ ├── __init__.py │ │ │ └── model.py │ │ ├── entrypoints │ │ │ ├── __init__.py │ │ │ └── flask_app.py ``` -------------------------------- ### Django ORM Model Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_02_repository.html This Django ORM example demonstrates model classes inheriting directly from `models.Model`, illustrating how Django models are coupled to the ORM. Django's ORM is known for being more restrictive than SQLAlchemy's. ```python class Order(models.Model): pass class OrderLine(models.Model): sku = models.CharField(max_length=255) qty = models.IntegerField() order = models.ForeignKey(Order) class Allocation(models.Model): ... ``` -------------------------------- ### Allocate Service with Precondition Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/appendix_validation.html The allocate service function demonstrates calling the `product_exists` precondition before proceeding with allocation logic. Ensures the product is available before attempting to allocate. ```python from allocation import ensure def allocate(event, uow): line = model.OrderLine(event.orderid, event.sku, event.qty) with uow: ensure.product_exists(event, uow) product = uow.products.get(line.sku) product.allocate(line) uow.commit() ``` -------------------------------- ### List Open Issues using Repository Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/blog/2017-09-13-commands-and-queries-handlers-and-views.html This Flask route demonstrates a simple way to list open issues by directly using repositories within the UI entrypoint. This approach is acceptable for simple queries but can lead to business logic mixing with glue code if not managed carefully. ```python @app.route("/issues") def list_issues(): with unit_of_work_manager.start() as unit_of_work: open_issues = unit_of_work.issues.find_by_status('open') return json.dumps(open_issues) ``` -------------------------------- ### Service Layer Test Example Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_05_high_gear_low_gear.html The same test logic rewritten to interact with the service layer, abstracting domain model details. ```python # service-layer test: def test_prefers_warehouse_batches_to_shipments(): in_stock_batch = Batch("in-stock-batch", "RETRO-CLOCK", 100, eta=None) shipment_batch = Batch("shipment-batch", "RETRO-CLOCK", 100, eta=tomorrow) repo = FakeRepository([in_stock_batch, shipment_batch]) session = FakeSession() line = OrderLine('oref', "RETRO-CLOCK", 10) services.allocate(line, repo, session) assert in_stock_batch.available_quantity == 90 assert shipment_batch.available_quantity == 100 ``` -------------------------------- ### FakeRepository Implementation Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_04_service_layer.html An in-memory repository implementation used for testing. It stores batches and provides methods to add, get, and list them. ```python class FakeRepository(repository.AbstractRepository): def __init__(self, batches): self._batches = set(batches) def add(self, batch): self._batches.add(batch) def get(self, reference): return next(b for b in self._batches if b.reference == reference) def list(self): return list(self._batches) ``` -------------------------------- ### Fake Repository for Testing Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html Demonstrates how to create a fake repository implementation for testing purposes. This allows for isolated unit tests without requiring a real database connection. ```python class FakeRepository(AbstractRepository): def __init__(self): self.entities = {} def add(self, entity): self.entities[entity.id] = entity def get(self, entity_id): return self.entities.get(entity_id) ``` -------------------------------- ### Fix Cross-References in HTML Source: https://context7.com/cosmicpython/cosmicpython.github.io/llms.txt Parses HTML content, finds links starting with '#', and updates them to point to the correct chapter files. ```Python def fix_xrefs(contents, chapter, chapter_info): """Fix cross-references to point to correct chapter files.""" parsed = html.fromstring(contents) links = parsed.cssselect(r'a[href^=\#]') for link in links: for other_chap in CHAPTERS: if other_chap == chapter: continue chapter_id = chapter_info[other_chap].href_id href = link.get('href') targets = ['#' + x for x in chapter_info[other_chap].xrefs] if href == '#' + chapter_id: link.set('href', f'/book/{other_chap}') elif href in targets: link.set('href', f'/book/{other_chap}{href}') return html.tostring(parsed) ``` -------------------------------- ### Calling Hello World Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/posts/2019-08-03-ioc-techniques.md The main script that imports and calls the 'hello_world' function. Demonstrates a typical execution flow. ```python # main.py from hello_world import hello_world if __name__ == "__main__": hello_world() ``` -------------------------------- ### IntEnum Definition and Usage Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/posts/2020-10-27-i-hate-enums.md An example using `IntEnum` which allows enums to duck-type as integers. Comparisons and conversions to integers work as expected. ```python class IBRAIN(IntEnum): SMALL = 1 MEDIUM = 2 GALAXY = 3 assert IBRAIN.SMALL == 1 assert int(IBRAIN.SMALL) == 1 assert IBRAIN.SMALL.value == 1 assert [thing for thing in IBRAIN] == [1, 2, 3] assert list(IBRAIN) == [1, 2, 3] assert [thing.value for thing in IBRAIN] == [1, 2, 3] assert random.choice(IBRAIN) in [1, 2, 3] # this still errors but: assert random.choice(list(IBRAIN)) in [1, 2, 3] # this is ok ``` -------------------------------- ### Configuring the Registry Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/posts/2019-08-03-ioc-techniques.md The main script that configures the 'OUTPUT_FUNCTION' in the 'hello_world.config' dictionary before calling 'hello_world'. ```python # main.py import hello_world hello_world.config["OUTPUT_FUNCTION"] = print if __name__ == "__main__": hello_world.hello_world() ``` -------------------------------- ### LightSwitch Class demonstrating CQS Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/blog/2017-09-13-commands-and-queries-handlers-and-views.html This class demonstrates Command Query Separation. The `is_on` property is a query, while `toggle_light` is a command with a side effect. It's recommended that commands do not return values to adhere strictly to CQS. ```python class LightSwitch: def toggle_light(self): self.light_is_on = not self.light_is_on return self.light_is_on @property def is_on(self): return self.light_is_on ``` -------------------------------- ### View Using Product Repository Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_12_cqrs.html This view retrieves product and batch information for a given order using a repository's helper method. It involves Python-based filtering after data retrieval. ```python from allocation import unit_of_work def allocations(orderid: str, uow: unit_of_work.AbstractUnitOfWork): with uow: products = uow.products.for_order(orderid=orderid) #(1) batches = [b for p in products for b in p.batches] #(2) return [ {'sku': b.sku, 'batchref': b.reference} for b in batches if orderid in b.orderids #(3) ] ``` -------------------------------- ### Integrate Precondition Checks in Service Layer Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/book.html Demonstrates how to call precondition checks within the service layer before performing core business logic. Ensures that necessary conditions are met before proceeding with operations like allocation. ```python from allocation import ensure def allocate(event, uow): line = mode.OrderLine(event.orderid, event.sku, event.qty) with uow: ensure.product_exists(uow, event) product = uow.products.get(line.sku) product.allocate(line) uow.commit() ``` -------------------------------- ### Examples of Value Objects Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/book/chapter_01_domain_model.html Demonstrates the creation and equality checks for different value object implementations: dataclass (Name), NamedTuple (Money), and namedtuple (Line). ```python from dataclasses import dataclass from typing import NamedTuple from collections import namedtuple @dataclass(frozen=True) class Name: first_name: str surname: str class Money(NamedTuple): currency: str value: int Line = namedtuple('Line', ['sku', 'qty']) ``` ```python assert Money('gbp', 10) == Money('gbp', 10) assert Name('Harry', 'Percival') != Name('Bob', 'Gregory') assert Line('RED-CHAIR', 5) == Line('RED-CHAIR', 5) ``` -------------------------------- ### Adding to a Global List Registry (Python) Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/posts/2019-08-03-ioc-techniques.md Demonstrates adding an item to a global list, simulating a simple registry pattern. This is typically done at application startup. ```python # john.py import hello_people hello_people.people.append("John") ``` ```python # martha.py import hello_people hello_people.people.append("Martha") ``` -------------------------------- ### Adapter Implementations for Ports Source: https://github.com/cosmicpython/cosmicpython.github.io/blob/master/posts/2017-09-08-repository-and-unit-of-work-pattern-in-python.md Provides example implementations for ReadablePort and WriteablePort interfaces. LightDetector reads light amplitude, Buzzer makes noise, Dial tracks current value, and Light controls an on/off state. ```python class LightDetector(ReadablePort): def read(self): return self.get_light_amplitude() class Buzzer(WriteablePort): def write(self, value): if value > 0: self.make_infuriating_noise() class Dial(ReadablePort): def read(self): return self.current_value class Light(self): def write(self, value): if value > 0: self.on = True else: self.on = False ```