# TurboBus
TurboBus is a powerful Python package designed to streamline the development of software applications adhering to the Command Responsibility Segregation (CRS) pattern.
## Installation
```bash
pip install turbobus
```
## Simple usage
Let's see an example using python typings. You can omit all the typing stuffs if you want to.
**God Mode ⚡**
```python3
from datetime import date
from turbobus.command import Command, CommandBus, CommandHandler, kw_only_frozen
from turbobus.constants import Provider
# We need to create a Command class that receives the values that the handler will use
# to execute the command. The Command class is a generic class that receives the return
# @kw_only_frozen: is a shortcut decorator for @dataclass(kw_only=True, frozen=True)
# Command[int]: is a generic class that receives a return_type.
# This is useful to check if the handler is returning the correct type
# And allow the CommandBus to know the return type of the command
@kw_only_frozen
class CalculateAgeCommand(Command[int]):
birthdate: str | date
# We need to create a CommandHandler class that will receive the Command class.
# The handler class must implement the execute method
# CommandHandler[CalculateAgeCommand]: is a generic class that receives the Command class
# this is useful to check if the handler is implementing the correct command class
class CalculateAgeHandler(CommandHandler[CalculateAgeCommand]):
# The execute method must receive the Command class and return
# the same type as in the Command class return_type
def execute(self, cmd: CalculateAgeCommand) -> int:
birthdate: date = cmd.birthdate if isinstance(cmd.birthdate, date) else date.fromisoformat(cmd.birthdate)
today = date.today()
age = today.year - birthdate.year - ((today.month, today.day) < (birthdate.month, birthdate.day))
return age
# We need to register the Command and Handler in the Provider
# This is necessary to allow the CommandBus to find the correct handler
# to execute the command
Provider.set(CalculateAgeCommand, CalculateAgeHandler)
if __name__ == '__main__':
# We need to create a CommandBus instance to execute the command
bus = CommandBus()
# Here we are executing the CalculateAgeCommand
# if you're using an IDE that supports type hinting
# you'll see that the result variable is inferred as int
# because the CalculateAgeCommand is a generic class
# that receives int as return_type
result = bus.execute(
CalculateAgeCommand(birthdate='1994-03-09')
)
print(f'You are {result} years old')
```
**Human Mode (No types, obviously 🙄)**
Here's the same example, but without types
```python3
from datetime import date
from turbobus.command import Command, CommandBus, CommandHandler, kw_only_frozen
from turbobus.constants import Provider
class CalculateAgeCommand(Command):
def __init__(self, birthdate):
self.birthdate = birthdate
class CalculateAgeHandler(CommandHandler):
def execute(self, cmd: CalculateAgeCommand):
birthdate = cmd.birthdate if isinstance(cmd.birthdate, date) else date.fromisoformat(cmd.birthdate)
today = date.today()
age = today.year - birthdate.year - ((today.month, today.day) < (birthdate.month, birthdate.day))
return age
Provider.set(CalculateAgeCommand, CalculateAgeHandler)
if __name__ == '__main__':
bus = CommandBus()
result = bus.execute(
CalculateAgeCommand(birthdate='1994-03-09')
)
print(f'You are {result} years old')
```
## Dependency injection
In many cases we're going to need to inject dependencies to our command handler. To accomplish that we have the `@inject` decorator. For example:
```python3
from abc import ABC, abstractmethod
from dataclasses import field
import uuid
from turbobus.command import Command, CommandBus, CommandHandler, kw_only_frozen
from turbobus.constants import Provider
from turbobus.injection import inject
# This is a simple Entity to represent a User
@kw_only_frozen
class UserEntity:
id: uuid.UUID = field(default_factory=uuid.uuid4)
name: str
email: str
# We need to define the repository interface
# to save and retrieve users
class UserRepository(ABC):
@abstractmethod
def get_by_id(self, id: uuid.UUID) -> UserEntity | None:
"""Get user by id"""
@abstractmethod
def save(self, user: UserEntity) -> None:
"""Save user"""
# This is an in-memory implementation of the UserRepository
class UserRepositoryInMemory(UserRepository):
def __init__(self):
self._users: dict[uuid.UUID, UserEntity] = {}
def get_by_id(self, id: uuid.UUID) -> UserEntity | None:
return self._users.get(id)
def save(self, user: UserEntity) -> None:
self._users[user.id] = user
# Let's create a command to create a user account
@kw_only_frozen
class CreateUserAccount(Command[None]):
name: str
email: str
# @inject is used to inject the dependencies
@inject
@kw_only_frozen
class CreateUserAccountHandler(CommandHandler[CreateUserAccount]):
user_repository: UserRepository
def execute(self, cmd: CreateUserAccount) -> None:
user = UserEntity(name=cmd.name, email=cmd.email)
# It's unnecessary to retrieve the user from the repository
# this is just to demonstrate that the user was saved
self.user_repository.save(user)
user = self.user_repository.get_by_id(user.id)
if user is None:
raise Exception('User not found')
print(f'Welcome {user.name}!')
Provider.set(UserRepository, UserRepositoryInMemory)
Provider.set(CreateUserAccount, CreateUserAccountHandler)
if __name__ == '__main__':
bus = CommandBus()
bus.execute(
CreateUserAccount(name='Christopher Flores', email='cafadev@outlook.com')
)
```
The `@inject` decorator also accepts the next parameters:
### Alias
The `@inject` decorator use the typing to resolve the required dependency. With the `alias: dict[str, Callable[..., Any]]` parameter you can specify a different implementation for the same interface. For example, let's say we have a UserRepository interface and then two different implementations; UserRepositoryInMemory and UserRepositorySQL.
```python3
Provider.set(UserRepository, UserRepositoryInMemory)
Provider.set('UserRepositorySQL', UserRepositorySQL)
@inject
class CreateUserAccount:
user: UserRepository
```
By default, the `@inject` will use the `UserRepositoryInMemory` to provide the dependency. Let's specify the `UserRepositorySQL` as the provider. To accomplish that we just need to specify the parameter name that we want to override, and then the Provider Key:
```python3
Provider.set(UserRepository, UserRepositoryInMemory)
Provider.set(UserRepositorySQL, UserRepositorySQL)
@inject(alias={ 'user': 'UserRepositorySQL' })
class CreateUserAccount:
user: UserRepository
```
### Only and Exclude
The `only` and `exclude` parameters in the `@inject` decorator allow you to fine-tune which dependencies should be resolved based on their names.
The `only` parameter specifies a list of dependency names that should exclusively be resolved. All other dependencies will be ignored.
```python3
@inject(only=['dependency1', 'dependency2'])
def my_function(dep1: Dependency1, dep2: Dependency2, dep3: Dependency3):
# Only dep1 and dep2 will be injected
pass
```
In this example, only dep1 and dep2 will be injected into the function because they are specified in the only list.
The `exclude` parameter specifies a list of dependency names that should be excluded from injection.
```python3
@inject(exclude=['dependency3'])
def my_function(dep1: Dependency1, dep2: Dependency2, dep3: Dependency3):
# Dep3 will not be injected
pass
```
In this example, dep3 will not be injected into the function because it is specified in the exclude list.
These parameters provide flexibility in controlling which dependencies are resolved, allowing you to customize the injection behavior according to your specific needs.
Raw data
{
"_id": null,
"home_page": "https://github.com/cafadev/turbobus",
"name": "turbobus",
"maintainer": null,
"docs_url": null,
"requires_python": null,
"maintainer_email": null,
"keywords": "command, bus, cqrs, crs, injection, ddd, domain, driven, design, pattern, python, turbobus",
"author": "Christopher A. Flores",
"author_email": "cafadev@outlook.com",
"download_url": "https://files.pythonhosted.org/packages/62/a2/394b2f70feed5f43edc1301bf10402a0959ce8ce9d43d20a8d8d3f9449a8/turbobus-1.0.0.tar.gz",
"platform": null,
"description": "# TurboBus\n\nTurboBus is a powerful Python package designed to streamline the development of software applications adhering to the Command Responsibility Segregation (CRS) pattern.\n\n## Installation\n```bash\npip install turbobus\n```\n\n## Simple usage\nLet's see an example using python typings. You can omit all the typing stuffs if you want to.\n\n**God Mode \u26a1**\n```python3\nfrom datetime import date\nfrom turbobus.command import Command, CommandBus, CommandHandler, kw_only_frozen\nfrom turbobus.constants import Provider\n\n# We need to create a Command class that receives the values that the handler will use\n# to execute the command. The Command class is a generic class that receives the return\n\n# @kw_only_frozen: is a shortcut decorator for @dataclass(kw_only=True, frozen=True)\n\n# Command[int]: is a generic class that receives a return_type.\n# This is useful to check if the handler is returning the correct type\n# And allow the CommandBus to know the return type of the command\n\n@kw_only_frozen \nclass CalculateAgeCommand(Command[int]):\n birthdate: str | date\n\n\n# We need to create a CommandHandler class that will receive the Command class.\n# The handler class must implement the execute method\n \n# CommandHandler[CalculateAgeCommand]: is a generic class that receives the Command class\n# this is useful to check if the handler is implementing the correct command class\nclass CalculateAgeHandler(CommandHandler[CalculateAgeCommand]):\n\n # The execute method must receive the Command class and return\n # the same type as in the Command class return_type\n def execute(self, cmd: CalculateAgeCommand) -> int:\n birthdate: date = cmd.birthdate if isinstance(cmd.birthdate, date) else date.fromisoformat(cmd.birthdate)\n\n today = date.today()\n age = today.year - birthdate.year - ((today.month, today.day) < (birthdate.month, birthdate.day))\n return age\n\n\n# We need to register the Command and Handler in the Provider\n# This is necessary to allow the CommandBus to find the correct handler\n# to execute the command\nProvider.set(CalculateAgeCommand, CalculateAgeHandler)\n\n\nif __name__ == '__main__':\n # We need to create a CommandBus instance to execute the command\n bus = CommandBus()\n\n # Here we are executing the CalculateAgeCommand\n # if you're using an IDE that supports type hinting\n # you'll see that the result variable is inferred as int\n # because the CalculateAgeCommand is a generic class\n # that receives int as return_type\n result = bus.execute(\n CalculateAgeCommand(birthdate='1994-03-09')\n )\n\n print(f'You are {result} years old')\n\n```\n\n**Human Mode (No types, obviously \ud83d\ude44)**\n\nHere's the same example, but without types\n```python3\nfrom datetime import date\nfrom turbobus.command import Command, CommandBus, CommandHandler, kw_only_frozen\nfrom turbobus.constants import Provider\n\n\nclass CalculateAgeCommand(Command):\n\n def __init__(self, birthdate):\n self.birthdate = birthdate\n\n\nclass CalculateAgeHandler(CommandHandler):\n\n def execute(self, cmd: CalculateAgeCommand):\n birthdate = cmd.birthdate if isinstance(cmd.birthdate, date) else date.fromisoformat(cmd.birthdate)\n\n today = date.today()\n age = today.year - birthdate.year - ((today.month, today.day) < (birthdate.month, birthdate.day))\n return age\n\nProvider.set(CalculateAgeCommand, CalculateAgeHandler)\n\nif __name__ == '__main__':\n bus = CommandBus()\n\n result = bus.execute(\n CalculateAgeCommand(birthdate='1994-03-09')\n )\n\n print(f'You are {result} years old')\n\n```\n\n## Dependency injection\nIn many cases we're going to need to inject dependencies to our command handler. To accomplish that we have the `@inject` decorator. For example:\n\n```python3\nfrom abc import ABC, abstractmethod\nfrom dataclasses import field\nimport uuid\nfrom turbobus.command import Command, CommandBus, CommandHandler, kw_only_frozen\nfrom turbobus.constants import Provider\nfrom turbobus.injection import inject\n\n\n# This is a simple Entity to represent a User\n@kw_only_frozen\nclass UserEntity:\n id: uuid.UUID = field(default_factory=uuid.uuid4)\n name: str\n email: str\n\n\n# We need to define the repository interface\n# to save and retrieve users\nclass UserRepository(ABC):\n\n @abstractmethod\n def get_by_id(self, id: uuid.UUID) -> UserEntity | None:\n \"\"\"Get user by id\"\"\"\n\n @abstractmethod\n def save(self, user: UserEntity) -> None:\n \"\"\"Save user\"\"\"\n\n\n# This is an in-memory implementation of the UserRepository\nclass UserRepositoryInMemory(UserRepository):\n\n def __init__(self):\n self._users: dict[uuid.UUID, UserEntity] = {}\n\n def get_by_id(self, id: uuid.UUID) -> UserEntity | None:\n return self._users.get(id)\n \n def save(self, user: UserEntity) -> None:\n self._users[user.id] = user\n\n\n# Let's create a command to create a user account\n@kw_only_frozen\nclass CreateUserAccount(Command[None]):\n name: str\n email: str\n\n\n# @inject is used to inject the dependencies\n@inject\n@kw_only_frozen\nclass CreateUserAccountHandler(CommandHandler[CreateUserAccount]):\n\n user_repository: UserRepository\n\n def execute(self, cmd: CreateUserAccount) -> None:\n user = UserEntity(name=cmd.name, email=cmd.email)\n\n # It's unnecessary to retrieve the user from the repository\n # this is just to demonstrate that the user was saved\n self.user_repository.save(user)\n user = self.user_repository.get_by_id(user.id)\n\n if user is None:\n raise Exception('User not found')\n \n print(f'Welcome {user.name}!')\n\n\nProvider.set(UserRepository, UserRepositoryInMemory)\nProvider.set(CreateUserAccount, CreateUserAccountHandler)\n\n\nif __name__ == '__main__':\n bus = CommandBus()\n\n bus.execute(\n CreateUserAccount(name='Christopher Flores', email='cafadev@outlook.com')\n )\n\n```\nThe `@inject` decorator also accepts the next parameters:\n\n### Alias\n\nThe `@inject` decorator use the typing to resolve the required dependency. With the `alias: dict[str, Callable[..., Any]]` parameter you can specify a different implementation for the same interface. For example, let's say we have a UserRepository interface and then two different implementations; UserRepositoryInMemory and UserRepositorySQL.\n\n```python3\nProvider.set(UserRepository, UserRepositoryInMemory)\nProvider.set('UserRepositorySQL', UserRepositorySQL)\n\n@inject\nclass CreateUserAccount:\n\n user: UserRepository\n```\n\nBy default, the `@inject` will use the `UserRepositoryInMemory` to provide the dependency. Let's specify the `UserRepositorySQL` as the provider. To accomplish that we just need to specify the parameter name that we want to override, and then the Provider Key:\n\n\n```python3\nProvider.set(UserRepository, UserRepositoryInMemory)\nProvider.set(UserRepositorySQL, UserRepositorySQL)\n\n@inject(alias={ 'user': 'UserRepositorySQL' })\nclass CreateUserAccount:\n\n user: UserRepository\n```\n\n### Only and Exclude\nThe `only` and `exclude` parameters in the `@inject` decorator allow you to fine-tune which dependencies should be resolved based on their names.\n\nThe `only` parameter specifies a list of dependency names that should exclusively be resolved. All other dependencies will be ignored.\n\n```python3\n@inject(only=['dependency1', 'dependency2'])\ndef my_function(dep1: Dependency1, dep2: Dependency2, dep3: Dependency3):\n # Only dep1 and dep2 will be injected\n pass\n```\n\nIn this example, only dep1 and dep2 will be injected into the function because they are specified in the only list.\n\nThe `exclude` parameter specifies a list of dependency names that should be excluded from injection.\n\n```python3\n@inject(exclude=['dependency3'])\ndef my_function(dep1: Dependency1, dep2: Dependency2, dep3: Dependency3):\n # Dep3 will not be injected\n pass\n\n```\n\nIn this example, dep3 will not be injected into the function because it is specified in the exclude list.\n\nThese parameters provide flexibility in controlling which dependencies are resolved, allowing you to customize the injection behavior according to your specific needs.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "TurboBus is a powerful Python package designed to streamline the development of software applications adhering to the Command Responsibility Segregation (CRS) pattern.",
"version": "1.0.0",
"project_urls": {
"Download": "https://github.com/cafadev/turbobus/releases/tag/v1.0.0",
"Homepage": "https://github.com/cafadev/turbobus"
},
"split_keywords": [
"command",
" bus",
" cqrs",
" crs",
" injection",
" ddd",
" domain",
" driven",
" design",
" pattern",
" python",
" turbobus"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "62a2394b2f70feed5f43edc1301bf10402a0959ce8ce9d43d20a8d8d3f9449a8",
"md5": "d7c1db239214e7823c6c6f8f08f2ef3c",
"sha256": "0a828da24a8e1a7f61c3d4f2cd70aaa908a19a65a852685018b520be4e9d8f20"
},
"downloads": -1,
"filename": "turbobus-1.0.0.tar.gz",
"has_sig": false,
"md5_digest": "d7c1db239214e7823c6c6f8f08f2ef3c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": null,
"size": 9257,
"upload_time": "2024-04-16T02:25:38",
"upload_time_iso_8601": "2024-04-16T02:25:38.388988Z",
"url": "https://files.pythonhosted.org/packages/62/a2/394b2f70feed5f43edc1301bf10402a0959ce8ce9d43d20a8d8d3f9449a8/turbobus-1.0.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-04-16 02:25:38",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "cafadev",
"github_project": "turbobus",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "turbobus"
}
JSON
