Skip to main content

TurboBus is an opinionated implementation of Command Responsibility Segregation pattern in python.

Project description

TurboBus

TurboBus is an opinionated implementation of Command Responsibility Segregation pattern in python.

Simple usage

Let's see an example using python typings. You can omit all the typing stuffs if you want to.

God Mode ⚡

from  dataclasses  import  dataclass
from  typing  import  TypeAlias

from  turbobus.bus  import  Command, CommandHandler, CommandBus
from  turbobus.decorators  import  command, injectable

LogHandlerType: TypeAlias  =  "ILogHandler"

@dataclass
class  LogCommand(Command[LogHandlerType]):
	content: str  


class  ILogHandler(CommandHandler[str]):
	...


@command(LogCommand)
class  LogHandler(ILogHandler):
	def  execute(self, cmd: LogCommand) -> str:
		return  cmd.content


if __name__ == '__main__':
	bus  =  CommandBus()

	result  =  bus.execute(
		LogCommand('Hello dude!')
	)
	print(result)  # Hello dude

Human Mode 🥱

from  dataclasses  import  dataclass

from  turbobus.bus  import  Command, CommandHandler
from  turbobus.decorators  import  command, injectable

@dataclass
class  LogCommand(Command):
	content


class  ILogHandler(CommandHandler):
	...

@command(LogCommand)
class  LogHandler(ILogHandler):
	def  execute(self, cmd: LogCommand) -> str:
		return  cmd.content


if __name__ == '__main__':
	bus  =  CommandBus()

	result  =  bus.execute(
		LogCommand('Hello dude!')
	)
	print(result)  # Hello dude

Dependency injection

In many cases we're going to need to inject dependencies to our command handler. To accomplish that we have two important tools: @injectable decorator and injecting function.

With the injectable decorator we can specify a class that is implementing the functionalities of the dependency. For example:

from turbobus.decorators import injectable
from log.axioma.log import ILogger


class ILogger(ABC):

    @abstractmethod
    def logger(self, text: str) -> None:
        ...


@injectable(ILogger)
class Logger:

    def logger(self, text: str) -> None:
        print('from logger', text)


@command(LogCommand)
@dataclass(kw_only=True)
class LogHandler(ILogHandler):

    logger = injecting(ILogger)

    def execute(self, cmd: LogCommand) -> str:
        if self.logger is not None:
            self.logger.logger(cmd.content)

        return cmd.content

As you can see in the example above, we're defining an abstract class with the logger method. Then we're doing the implementation of the ILogger and we're indicating that in the @injectable(ILogger).

Then, using the injecting function, TurboBus is going to map that dependency and inject the instance in the attribute.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

turbobus-1.0.0a3.tar.gz (3.2 kB view details)

Uploaded Source

File details

Details for the file turbobus-1.0.0a3.tar.gz.

File metadata

  • Download URL: turbobus-1.0.0a3.tar.gz
  • Upload date:
  • Size: 3.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? Yes
  • Uploaded via: twine/4.0.2 CPython/3.11.6

File hashes

Hashes for turbobus-1.0.0a3.tar.gz
Algorithm Hash digest
SHA256 6a1fa2483d01b544f2c222b737c1d53d07f3df8a337cc72793b47805943c91b5
MD5 40f6832b139a08c5e617a2c7f25b2354
BLAKE2b-256 b58ca8c9996dcc4d17253c2384c23ca30074326e92bcafea7ccb1a482e68c3fa

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page