| Name | command-system JSON |
| Version |
0.2.0
JSON |
| download |
| home_page | None |
| Summary | Type-safe command pattern implementation in pure Python |
| upload_time | 2025-07-13 02:55:34 |
| maintainer | None |
| docs_url | None |
| author | Alec Zaiane |
| requires_python | >=3.11 |
| license | None |
| keywords |
command
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
> Parts of this readme do not render correctly on pypi, see the [repository](https://github.com/alec-zaiane/command-system) for a better experience
# Command System
A type-safe implementation of the command pattern in Python, designed to handle command execution with a lifecycle that includes deferral, cancellation, and dependency management.
With this package, you can create commands that can be queued and manage their own lifecycle, allowing for interacting with external systems, or building your own systems with complex interactions in a safe, clean, and maintainable way.
## Example Usage
Simply subclass `Command`, `CommandArgs`, and optionally `CommandResponse` to create a custom command for your use case.
```python
# SayHelloCommand.py
from command_system import (
Command,
CommandArgs,
CommandResponse,
ExecutionResponse,
)
from dataclasses import dataclass
@dataclass
class SayHelloArgs(CommandArgs):
name: str
@dataclass
class SayHelloResponse(CommandResponse):
message: str = ""
class SayHelloCommand(Command[SayHelloArgs, SayHelloResponse]):
ARGS = SayHelloArgs
_response_type = SayHelloResponse
def execute(self) -> ExecutionResponse:
if not self.args.name:
return ExecutionResponse.failure("Name cannot be empty.")
self.response.message = f"Hello, {self.args.name}!"
return ExecutionResponse.success()
```
You can then use the `CommandQueue` to submit commands, and then `queue.process_once()` or `queue.process_all()` to execute them.
```python
# main.py
from command_system import CommandQueue
from SayHelloCommand import SayHelloCommand, SayHelloArgs
queue = CommandQueue()
response = queue.submit(SayHelloCommand(SayHelloArgs(name="Alice")))
print(response.status) # Pending
queue.process_once()
print(response.status) # Completed
print(response.message) # Hello, Alice!
```
## Command Lifecycle
```mermaid
flowchart TD
A[create command] -->|"queue.submit(command)"| B[responseStatus.CREATED]
B --> C{"command.should_defer()"}
C -->|"DeferResponse.defer()"| W["responseStatus.PENDING"]
W --> C
C -->|"DeferResponse.proceed()"| D{"command.should_cancel()"}
D -->|"CancelResponse.cancel()"| E[ResponseStatus.CANCELED]
D -->|"CancelResponse.proceed()"| F["command.execute()"]
F -->|"ExecutionResponse.success()"| G["ResponseStatus.COMPLETED"]
F -->|"ExecutionResponse.failure()"| H["ResponseStatus.FAILED"]
```
## Chaining Commands
You can chain commands together using the `CommandChainBuilder`. By doing this, you can create a sequence of commands that will be executed in order, passing the output of one command as the input to the next (with customizable transformations).
```python
from
from my_command import MyCommand, MyCommandArgs
from my_other_command import MyOtherCommand, MyOtherCommandArgs
queue = CommandQueue()
chain = CommandChainBuilder[str, str].start(
"my initial input",
args_factory=lambda x: MyCommandArgs(input_data=x),
command_class=MyCommand,
result_extractor=lambda response: response.output_data,
).then(
args_factory=lambda x: MyOtherCommandArgs(input_data=x, other_data="another value"),
command_class=MyOtherCommand,
result_extractor=lambda response: response.output_data,
).build(queue)
queue.submit(chain)
queue.process_all()
```
## Dependency Management
Commands can now define dependencies on other commands. Dependencies are evaluated before each lifecycle check begins, and they can preemptively defer or cancel the command based on their statuses.
Dependencies only have an effect when using the CommandQueue to process commands. If you write your own command processing logic, you will need to handle dependencies manually.
**It is your responsibility to ensure that dependencies do not create circular references that will cause an infinite loop.**
### Adding Dependencies
You can add dependencies to a command by passing them during initialization or using the `add_dependency` method. Dependencies can be other commands or wrapped in `DependencyEntry` for more control.
```python
from command_system import Command, CommandArgs, DependencyEntry
from MyCommand import MyCommand
# Example usage
dependency_command = MyCommand(MyCommand.ARGS())
main_command = MyCommand(MyCommand.ARGS(), dependencies=[dependency_command])
```
### Writing Dependency Rules
When defining dependencies, you can specify actions for different statuses of the dependent command:
- `on_pending`: What to do if the dependency is pending/created (default: "defer").
- `on_canceled`: What to do if the dependency is canceled (default: "cancel").
- `on_failed`: What to do if the dependency fails (default: "cancel").
- `on_completed`: What to do if the dependency is completed (default: "proceed").
### Example
```python
from command_system import CommandQueue, DependencyEntry
from MyCommand import MyCommand
queue = CommandQueue()
dependency_command = MyCommand(MyCommand.ARGS(...))
main_command = MyCommand(MyCommand.ARGS(...), dependencies=[DependencyEntry(dependency_command, on_pending="cancel")])
queue.submit(dependency_command)
queue.submit(main_command)
queue.process_once()
```
In this example, `main_command` will cancel if `dependency_command` is still pending.
## Creating a command
Subclass `CommandArgs` and add any arguments your command needs. This class will be used to pass parameters to your command.
> [!CAUTION]
> Don't add your arguments directly to the `Command` class. The args class is required for command chaining to work in a type-safe manner.
Subclass `Command` and set the `ARGS` class attribute to the subclass of `CommandArgs` you created. Implement the `execute` method to define the command's behavior.
- You can also override `should_defer` and `should_cancel` methods to control the command's lifecycle.
Optionally, you can create a custom response class by subclassing `CommandResponse` so that your command can return specific type-safe data. If you do this, set the `_response_type` class attribute of your `Command` subclass to your custom response class.
### Writing your `execute` method
> [!WARNING]
> Your `execute` method should not return your custom response class directly.
> The `self.response` attribute is automatically set to an instance of your custom response class, which you should *modify* instead. Then, return an `ExecutionResponse` instance to indicate the command's success or failure.
### Writing `should_defer` and `should_cancel` methods
These methods can be overridden to control the command's lifecycle. They must return a `DeferResponse` or `CancelResponse` instance, respectively. You can use them to set conditions for deferring or canceling the command.
### Complex Command example
For an example of deferring and canceling commands, see the [tests/test_defer_cancel.py](tests/test_defer_cancel.py) file.
Raw data
{
"_id": null,
"home_page": null,
"name": "command-system",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "command",
"author": "Alec Zaiane",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/05/25/beb0d8ff76c100ead77c393b880814862059571f802416bb01867eaa935d/command_system-0.2.0.tar.gz",
"platform": null,
"description": "> Parts of this readme do not render correctly on pypi, see the [repository](https://github.com/alec-zaiane/command-system) for a better experience\n\n# Command System\nA type-safe implementation of the command pattern in Python, designed to handle command execution with a lifecycle that includes deferral, cancellation, and dependency management.\n\nWith this package, you can create commands that can be queued and manage their own lifecycle, allowing for interacting with external systems, or building your own systems with complex interactions in a safe, clean, and maintainable way.\n\n\n## Example Usage\nSimply subclass `Command`, `CommandArgs`, and optionally `CommandResponse` to create a custom command for your use case.\n\n```python\n# SayHelloCommand.py\nfrom command_system import (\n Command,\n CommandArgs,\n CommandResponse,\n ExecutionResponse,\n)\nfrom dataclasses import dataclass\n\n@dataclass\nclass SayHelloArgs(CommandArgs):\n name: str\n\n@dataclass\nclass SayHelloResponse(CommandResponse):\n message: str = \"\"\n\nclass SayHelloCommand(Command[SayHelloArgs, SayHelloResponse]):\n ARGS = SayHelloArgs\n _response_type = SayHelloResponse\n\n def execute(self) -> ExecutionResponse:\n if not self.args.name:\n return ExecutionResponse.failure(\"Name cannot be empty.\")\n self.response.message = f\"Hello, {self.args.name}!\"\n return ExecutionResponse.success()\n```\n\nYou can then use the `CommandQueue` to submit commands, and then `queue.process_once()` or `queue.process_all()` to execute them.\n\n```python\n# main.py\nfrom command_system import CommandQueue\nfrom SayHelloCommand import SayHelloCommand, SayHelloArgs\n\nqueue = CommandQueue()\nresponse = queue.submit(SayHelloCommand(SayHelloArgs(name=\"Alice\")))\nprint(response.status) # Pending\nqueue.process_once()\nprint(response.status) # Completed\nprint(response.message) # Hello, Alice!\n```\n\n## Command Lifecycle\n```mermaid\nflowchart TD\n A[create command] -->|\"queue.submit(command)\"| B[responseStatus.CREATED]\n B --> C{\"command.should_defer()\"}\n C -->|\"DeferResponse.defer()\"| W[\"responseStatus.PENDING\"]\n W --> C\n C -->|\"DeferResponse.proceed()\"| D{\"command.should_cancel()\"}\n D -->|\"CancelResponse.cancel()\"| E[ResponseStatus.CANCELED]\n D -->|\"CancelResponse.proceed()\"| F[\"command.execute()\"]\n F -->|\"ExecutionResponse.success()\"| G[\"ResponseStatus.COMPLETED\"]\n F -->|\"ExecutionResponse.failure()\"| H[\"ResponseStatus.FAILED\"]\n```\n## Chaining Commands\nYou can chain commands together using the `CommandChainBuilder`. By doing this, you can create a sequence of commands that will be executed in order, passing the output of one command as the input to the next (with customizable transformations).\n\n```python\nfrom \nfrom my_command import MyCommand, MyCommandArgs\nfrom my_other_command import MyOtherCommand, MyOtherCommandArgs\n\nqueue = CommandQueue()\nchain = CommandChainBuilder[str, str].start(\n \"my initial input\",\n args_factory=lambda x: MyCommandArgs(input_data=x),\n command_class=MyCommand,\n result_extractor=lambda response: response.output_data,\n).then(\n args_factory=lambda x: MyOtherCommandArgs(input_data=x, other_data=\"another value\"),\n command_class=MyOtherCommand,\n result_extractor=lambda response: response.output_data,\n).build(queue)\n\nqueue.submit(chain)\nqueue.process_all()\n```\n\n## Dependency Management\nCommands can now define dependencies on other commands. Dependencies are evaluated before each lifecycle check begins, and they can preemptively defer or cancel the command based on their statuses.\n\nDependencies only have an effect when using the CommandQueue to process commands. If you write your own command processing logic, you will need to handle dependencies manually.\n\n**It is your responsibility to ensure that dependencies do not create circular references that will cause an infinite loop.**\n\n### Adding Dependencies\nYou can add dependencies to a command by passing them during initialization or using the `add_dependency` method. Dependencies can be other commands or wrapped in `DependencyEntry` for more control.\n\n```python\nfrom command_system import Command, CommandArgs, DependencyEntry\nfrom MyCommand import MyCommand\n\n# Example usage\ndependency_command = MyCommand(MyCommand.ARGS())\nmain_command = MyCommand(MyCommand.ARGS(), dependencies=[dependency_command])\n```\n\n### Writing Dependency Rules\nWhen defining dependencies, you can specify actions for different statuses of the dependent command:\n- `on_pending`: What to do if the dependency is pending/created (default: \"defer\").\n- `on_canceled`: What to do if the dependency is canceled (default: \"cancel\").\n- `on_failed`: What to do if the dependency fails (default: \"cancel\").\n- `on_completed`: What to do if the dependency is completed (default: \"proceed\").\n\n### Example\n```python\nfrom command_system import CommandQueue, DependencyEntry\nfrom MyCommand import MyCommand\n\nqueue = CommandQueue()\ndependency_command = MyCommand(MyCommand.ARGS(...))\nmain_command = MyCommand(MyCommand.ARGS(...), dependencies=[DependencyEntry(dependency_command, on_pending=\"cancel\")])\n\nqueue.submit(dependency_command)\nqueue.submit(main_command)\nqueue.process_once()\n```\n\nIn this example, `main_command` will cancel if `dependency_command` is still pending.\n\n## Creating a command\nSubclass `CommandArgs` and add any arguments your command needs. This class will be used to pass parameters to your command.\n\n> [!CAUTION]\n> Don't add your arguments directly to the `Command` class. The args class is required for command chaining to work in a type-safe manner.\n\nSubclass `Command` and set the `ARGS` class attribute to the subclass of `CommandArgs` you created. Implement the `execute` method to define the command's behavior.\n- You can also override `should_defer` and `should_cancel` methods to control the command's lifecycle.\n\nOptionally, you can create a custom response class by subclassing `CommandResponse` so that your command can return specific type-safe data. If you do this, set the `_response_type` class attribute of your `Command` subclass to your custom response class.\n\n### Writing your `execute` method\n> [!WARNING] \n> Your `execute` method should not return your custom response class directly. \n> The `self.response` attribute is automatically set to an instance of your custom response class, which you should *modify* instead. Then, return an `ExecutionResponse` instance to indicate the command's success or failure. \n\n### Writing `should_defer` and `should_cancel` methods\nThese methods can be overridden to control the command's lifecycle. They must return a `DeferResponse` or `CancelResponse` instance, respectively. You can use them to set conditions for deferring or canceling the command.\n\n### Complex Command example\nFor an example of deferring and canceling commands, see the [tests/test_defer_cancel.py](tests/test_defer_cancel.py) file.",
"bugtrack_url": null,
"license": null,
"summary": "Type-safe command pattern implementation in pure Python",
"version": "0.2.0",
"project_urls": {
"Repository": "https://github.com/alec-zaiane/command-system/"
},
"split_keywords": [
"command"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "a4f5105263ee9ad6958284a04c666b836592aa738a5295637a0fe4d9be7b1974",
"md5": "6a8821ef00512cd7c74d0d120ea11deb",
"sha256": "ed814bcf5684f7c12210a35121f620268bf2542ae3bc02201cc933b1448c7fb5"
},
"downloads": -1,
"filename": "command_system-0.2.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "6a8821ef00512cd7c74d0d120ea11deb",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 14956,
"upload_time": "2025-07-13T02:55:33",
"upload_time_iso_8601": "2025-07-13T02:55:33.109157Z",
"url": "https://files.pythonhosted.org/packages/a4/f5/105263ee9ad6958284a04c666b836592aa738a5295637a0fe4d9be7b1974/command_system-0.2.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "0525beb0d8ff76c100ead77c393b880814862059571f802416bb01867eaa935d",
"md5": "86416f34fcd69ce48aee6f0ea1e0adf1",
"sha256": "2a8f27c37bba72a799bed0d1b67be38a3072348573cffc8b8d6b02990617aab0"
},
"downloads": -1,
"filename": "command_system-0.2.0.tar.gz",
"has_sig": false,
"md5_digest": "86416f34fcd69ce48aee6f0ea1e0adf1",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 11596,
"upload_time": "2025-07-13T02:55:34",
"upload_time_iso_8601": "2025-07-13T02:55:34.147733Z",
"url": "https://files.pythonhosted.org/packages/05/25/beb0d8ff76c100ead77c393b880814862059571f802416bb01867eaa935d/command_system-0.2.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-13 02:55:34",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "alec-zaiane",
"github_project": "command-system",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "command-system"
}
