Connect lending and catalogue domains over Redis broker

The catalogue domain now publishes a message over channel "book_instance_added"
which is consumed by the lending domain to initialize the local `Book` aggregate.

Makefile

@@ -1,5 +1,5 @@
 test:
-	pytest
+	pytest tests/lending tests/catalogue
 
 test-cov:
-	pytest --cov=src/lending --cov-report=term-missing --cov-branch
+	pytest tests/lending tests/catalogue --cov=src/lending --cov=src/catalogue --cov-report=term-missing --cov-branch

README.md

@@ -6,7 +6,8 @@ Protean example implementation
 |-----------------|-------------------------------------------------------------------------|
 | Source          | [DDD by Examples - Library](https://github.com/ddd-by-examples/library) |
 | Pattern         | CQRS                                                                    |
-| Protean Version | 0.12.0                                                                  |
+| Protean Version | 0.12.1                                                                  |
+| Build Status    | [Build Status](https://github.com/proteanhq/library-cqrs/actions/workflows/ci.yml/badge.svg) |
 | Coverage        | [![codecov](https://codecov.io/github/proteanhq/library-cqrs/graph/badge.svg?token=onIFcl4Dg5)](https://codecov.io/github/proteanhq/library-cqrs)|
 
 ## Contributing

docker-compose.yml

@@ -6,6 +6,13 @@ services:
     ports:
       - 5433:5432
 
+  redis:
+    image: redis:7.0.11
+    ports:
+      - "6379:6379"
+    environment:
+      - ALLOW_EMPTY_PASSWORD=yes
+
 volumes:
   db-data:
     driver: local

pyproject.toml

@@ -19,9 +19,11 @@ packages = [
 [tool.poetry.dependencies]
 python = "^3.11"
 # protean = {version = "0.12.1", extras=[] }
-protean = { git = "https://github.com/proteanhq/protean.git", branch = "main" }
+protean = { git = "https://github.com/proteanhq/protean.git", branch = "main" , extras=["postgresql", "sqlite", "message_db", "redis"]}
+# protean = { path = "../../protean", develop=true}
 sqlalchemy = "^2.0.31"
 fastapi = "^0.111.1"
+redis = "^5.0.7"
 
 [tool.poetry.group.test]
 optional = true

src/catalogue/main.py

@@ -1,5 +1,7 @@
+import json
 from typing import Annotated
 
+import redis
 from fastapi import Depends, FastAPI
 from pydantic import BaseModel, Field
 from sqlalchemy.orm import Session
@@ -14,6 +16,10 @@
 Base.metadata.create_all(bind=engine)
 
 
+# Redis setup
+redis_client = redis.Redis(host="localhost", port=6379, db=0)
+
+
 def get_db():
     db = SessionLocal()
     try:
@@ -56,4 +62,17 @@ async def add_book_instance(
 
     db.add(new_book_instance)
     db.commit()
+
+    # Raise Event
+    event_dict = {
+        "instance_id": new_book_instance.id,
+        "isbn": new_book_instance.isbn,
+        "title": new_book_instance.book.title,
+        "summary": new_book_instance.book.summary,
+        "price": new_book_instance.book.price,
+        "is_circulating": new_book_instance.is_circulating,
+        "added_at": new_book_instance.added_at.isoformat(),
+    }
+    redis_client.publish("book_instance_added", json.dumps(event_dict))
+
     return {"message": "Book instance added successfully"}

src/catalogue/models.py

@@ -1,4 +1,15 @@
-from sqlalchemy import Boolean, Column, Float, ForeignKey, Integer, String, Text
+from sqlalchemy import (
+    Boolean,
+    Column,
+    DateTime,
+    Float,
+    ForeignKey,
+    Integer,
+    String,
+    Text,
+    func,
+)
+from sqlalchemy.orm import relationship
 
 from .database import Base
 
@@ -11,10 +22,15 @@ class Book(Base):
     summary = Column(Text)
     price = Column(Float, nullable=False)
 
+    instances = relationship("BookInstance", back_populates="book")
+
 
 class BookInstance(Base):
     __tablename__ = "book_instances"
 
     id = Column(Integer, primary_key=True, index=True, autoincrement=True)
     isbn = Column(String(13), ForeignKey("books.isbn"), nullable=False)
     is_circulating = Column(Boolean, default=True)
+    added_at = Column(DateTime, server_default=func.now())
+
+    book = relationship("Book", back_populates="instances")

src/lending/domain.toml

@@ -9,15 +9,28 @@ command_processing = "sync"
 [databases.default]
 provider = "memory"
 
+[databases.production]
+provider = "postgresql"
+database_uri = "postgresql://postgres:postgres@localhost:5432/postgres"
+
 [databases.memory]
 provider = "memory"
 
 [brokers.default]
 provider = "inline"
 
+[brokers.production]
+provider = "redis"
+URI = "redis://127.0.0.1:6379/0"
+IS_ASYNC = true
+
 [caches.default]
 provider = "memory"
 
+[event_store]
+provider = "message_db"
+database_uri = "postgresql://message_store@localhost:5433/message_store"
+
 [custom]
 CHECKOUT_PERIOD = 60  # Days
 HOLD_EXPIRY_DAYS = 7  # Days

src/lending/model/book.py

@@ -1,6 +1,6 @@
 from enum import Enum
 
-from protean import handle
+from protean import UnitOfWork, current_domain, handle
 from protean.fields import String
 
 from lending.domain import lending
@@ -34,6 +34,12 @@ class Book:
     )
 
 
+@lending.repository(part_of=Book)
+class BookRepository:
+    def find_by_isbn(self, isbn):
+        return current_domain.repository_for(Book)._dao.find_by(isbn=isbn)
+
+
 @lending.event_handler(part_of=Book, stream_category="library::patron")
 class PatronHoldEventsHandler:
     @handle(HoldPlaced)
@@ -44,3 +50,22 @@ def mark_book_on_hold(self, event: HoldPlaced):
         book.status = BookStatus.ON_HOLD.value
 
         repo.add(book)
+
+
+@lending.subscriber(channel="book_instance_added")
+class AddBookToLibrary:
+    def __call__(self, message: dict):
+        with UnitOfWork():
+            repo = lending.repository_for(Book)
+
+            book_type = (
+                BookType.CIRCULATING.value
+                if message["is_circulating"]
+                else BookType.RESTRICTED.value
+            )
+            book = Book(
+                isbn=message["isbn"],
+                book_type=book_type,
+                status=BookStatus.AVAILABLE.value,
+            )
+            repo.add(book)

src/shared/events/__init__.py

@@ -0,0 +1,3 @@
+from .book_instance_added import BookInstanceAdded
+
+__all__ = ["BookInstanceAdded"]

src/shared/events/book_instance_added.py

@@ -0,0 +1,12 @@
+from protean import BaseEvent
+from protean.fields import Boolean, DateTime, Float, Identifier, String, Text
+
+
+class BookInstanceAdded(BaseEvent):
+    instance_id = Identifier(required=True)
+    isbn = String(required=True)
+    title = String(required=True)
+    summary = Text(required=True)
+    price = Float(required=True)
+    is_circulating = Boolean(required=True)
+    added_at = DateTime(required=True)

tests/catalogue/bdd/test_get_db.py

@@ -0,0 +1,10 @@
+from catalogue.main import get_db
+from sqlalchemy.orm import Session
+
+
+def test_get_db():
+    db = next(get_db())
+    assert db is not None
+    assert isinstance(db, Session)
+    assert db.bind is not None
+    assert db.bind.url.database == "./books.db"

tests/lending/bdd/additions/features/book_instanced_added.feature

@@ -0,0 +1,9 @@
+Feature: Check Out a Book
+
+  Scenario: Book instance is added to catalogue
+    Given the librarian added a CIRCULATING book instance
+    Then a CIRCULATING book instance is successfully added as available
+
+  Scenario: Restricted Book instance is added to catalogue
+    Given the librarian added a RESTRICTED book instance
+    Then a RESTRICTED book instance is successfully added as available