Petisco is a framework for helping Python developers to build clean Applications
Project description
petisco 🍪
Petisco is a framework for helping Python developers to build clean Applications in Python.
⚠️ Disclaimer: Current version now is v1 (not stable yet).
Find deprecated version (v0) in legacy
branch.
Installation 💻
pip install petisco
Installation with Extras
pip install petisco[fastapi,sqlalchemy,elastic,rabbitmq,slack,redis]
Getting Started 📈
Model your Domain
petisco 🍪
is a framework which helps you to model you domain.
Imagine you have a domain where your business logic is to manage a task system, let's see some examples that can help you.
UUID
Use Uuid
to generate new identifiers.
from petisco import Uuid
uuid = Uuid.v4()
Additionally, you can extend it:
from petisco import Uuid
class TaskId(Uuid):
pass
task_id = TaskId.v4()
Domain Event
Use DomainEvent
to explicitly implement side effects of changes within your domain.
from petisco import DomainEvent, Uuid
class TaskId(Uuid):
pass
class TaskCreated(DomainEvent):
task_id: TaskId
class TaskRemoved(DomainEvent):
task_id: TaskId
class TaskRetrieved(DomainEvent):
task_id: TaskId
DomainEvent
inherits from Message
that have available dict
and json
method to encode the info.
my_domain_event = TaskCreated(task_id=TaskId.v4())
print(my_domain_event.json())
The result should be something like the following:
{"data": {"id": "3a4d78aa-6870-41cb-aa14-964831511d86", "type": "task.created", "type_message": "domain_event", "version": 1, "occurred_on": "2021-12-28 14:11:47.845618", "attributes": {"task_id": "a7f8b62a-c9e5-4f3c-a451-47cd1965958f"}, "meta": {}}}
If you use CQRS you can use also the Command
class.
from petisco import Command, Uuid
class TaskId(Uuid):
pass
class UpdateTask(Command):
task_id: TaskId
my_command = UpdateTask(task_id=TaskId.v4())
print(my_command.json())
The result:
{"data": {"id": "1f35e414-0636-4983-987e-13d522749709", "type": "update.task", "type_message": "command", "version": 1, "occurred_on": "2021-12-28 14:19:09.149651", "attributes": {"task_id": "db0970be-f6b6-478b-976a-f83e85112b90"}, "meta": {}}}
Domain Error
from petisco import DomainError
class TaskAlreadyExistError(DomainError):
pass
class TaskNotFoundError(DomainError):
pass
⚠️ TODO: Add documentation on how to specify the message
Value Objects
Extend ValueObject
to model your Value Objects.
from pydantic import validator
from meiga import Failure
from petisco import ValueObject, DomainError
class EmptyValueObjectError(DomainError): pass
class ExceedLengthLimitValueObjectError(DomainError): pass
def ensure_not_empty_value(value, classname: str = None):
if value is None:
raise Failure(EmptyValueObjectError(classname))
def ensure_value_is_less_than_200_char(value):
if len(value) > 200:
raise ExceedLengthLimitValueObjectError(value)
class Description(ValueObject):
@validator('value')
def validate_value(cls, value):
ensure_not_empty_value(value, cls.__name__)
ensure_value_is_less_than_200_char(value)
return value.title()
Aggregate Root
Extend AggregateRoot
to model your Aggregate Roots
from petisco import AggregateRoot
from datetime import datetime
from app.src import Description
from app.src.tasks.shared.domain.events import TaskCreated
from app.src.tasks.shared.domain.task_id import TaskId
from app.src.tasks.shared.domain.title import Title
class Task(AggregateRoot):
title: Title
description: Description
created_at: datetime
@staticmethod
def create(task_id: TaskId, title: Title, description: Description):
task = Task(aggregate_id=task_id, title=title, description=description, created_at=datetime.utcnow())
task.record(TaskCreated(task_id=task_id))
return task
⚠️ TODO: How we can take advantage of aggregate root (pull_domain_events, record)
Controller
Use Controller
class to define and configure inputs and outputs or your entry point.
You can use a simpler and default configuration
from petisco Controller
import random
class MyController(Controller):
def execute(self) -> bool:
return random.choice([True, False])
Or define some configurations using the inner class Config
from petisco DomainError, Controller, PrintMiddleware
import random
class MyError(DomainError): ...
class MyController(Controller):
class Config:
middlewares = [PrintMiddleware]
success_handler = lambda result: {"message": f"MyController set {result}"}
error_map = {MyError: {"message": "something wrong happens"}}
def execute(self) -> bool:
return random.choice([True, False])
If you want to set a default middleware for every Controller, you can use the envvar PETISCO_DEFAULT_MIDDLEWARES
:
PETISCO_DEFAULT_MIDDLEWARES=PrintMiddleware
: to configure PrintMiddlewarePETISCO_DEFAULT_MIDDLEWARES=NotifierMiddleware
: to configure NotifierMiddlewarePETISCO_DEFAULT_MIDDLEWARES=PrintMiddleware,NotifierMiddleware
: to configure several middlewares
Messages
⚠️ TODO: How we use the Message Manager!
Testing :white_check_mark:
Extras
RabbitMQ
To test how petisco can help you on rabbitmq queues management you need to run locally a RabbitMQ application, otherwise related test will be skipped. Please, check the official doc here: https://www.rabbitmq.com/download.html.
Run RabbitMQ with docker
docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management
You can check the RabbitMQ status on http://localhost:15672/#/ (guest:guest).
Find here some petisco examples.
Configure the rabbitmq
python examples/rabbitmq/configure.py
Start consuming messages
python examples/rabbitmq/consume.py
Publish some Domain Events
python examples/rabbitmq/publish_domain_events.py
Dispatch a Command
python examples/rabbitmq/dispatch_commands.py
Configurations
RABBITMQ_HEARTBEAT
: (default: 60 s)RABBITMQ_USER
: (default: guest)RABBITMQ_PASSWORD
: (default: guest)RABBITMQ_HOST
: (default: localhost)RABBITMQ_HOST
: (default: 5672)RABBITMQ_CONNECTION_NUM_MAX_RETRIES
: (default: 15)RABBITMQ_CONNECTION_WAIT_SECONDS_RETRY
: (default: 1)RABBITMQ_MESSAGE_TTL
: (default 1000 ms) If a queue is already created it will generate a precodition failure.
Development
Using lume
pip install lume
Then:
lume -install -all
Contact 📬
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.