Skip to content

logo

azure-devops-builds azure-devops-coverage readthedocs gitter pypi


The business transaction DSL

stories is a business transaction DSL. It provides a simple way to define a complex business transaction that includes processing over many steps and by many different objects. It makes error handling a primary concern by taking a “Railway Oriented Programming” approach to capturing and returning errors from any step in the transaction.

stories is based on the following ideas:

  • A business transaction is a series of operations where any can fail and stop the processing.
  • A business transaction can describe its steps on an abstract level without being coupled to any details about how individual operations work.
  • A business transaction doesn’t have any state.
  • Each operation shouldn’t accumulate state, instead it should receive an input and return an output without causing any side-effects.
  • The only interface of an operation is ctx.
  • Each operation provides a meaningful piece of functionality and can be reused.
  • Errors in any operation should be easily caught and handled as part of the normal application flow.

Example

stories provide a simple way to define a complex business scenario that include many processing steps.

>>> from stories import story, arguments, Success, Failure, Result
>>> from app.repositories import load_category, load_profile, create_subscription

>>> class Subscribe:
...
...     @story
...     @arguments('category_id', 'profile_id')
...     def buy(I):
...
...         I.find_category
...         I.find_profile
...         I.check_balance
...         I.persist_subscription
...         I.show_subscription
...
...     def find_category(self, ctx):
...
...         ctx.category = load_category(ctx.category_id)
...         return Success()
...
...     def find_profile(self, ctx):
...
...         ctx.profile = load_profile(ctx.profile_id)
...         return Success()
...
...     def check_balance(self, ctx):
...
...         if ctx.category.cost < ctx.profile.balance:
...             return Success()
...         else:
...             return Failure()
...
...     def persist_subscription(self, ctx):
...
...         ctx.subscription = create_subscription(category=ctx.category, profile=ctx.profile)
...         return Success()
...
...     def show_subscription(self, ctx):
...
...         return Result(ctx.subscription)

>>> Subscribe().buy(category_id=1, profile_id=1)
Subscription(primary_key=10)
>>> import asyncio
>>> from stories import story, arguments, Success, Failure, Result
>>> from aioapp.repositories import load_category, load_profile, create_subscription

>>> class Subscribe:
...
...     @story
...     @arguments('category_id', 'profile_id')
...     def buy(I):
...
...         I.find_category
...         I.find_profile
...         I.check_balance
...         I.persist_subscription
...         I.show_subscription
...
...     async def find_category(self, ctx):
...
...         ctx.category = await load_category(ctx.category_id)
...         return Success()
...
...     async def find_profile(self, ctx):
...
...         ctx.profile = await load_profile(ctx.profile_id)
...         return Success()
...
...     async def check_balance(self, ctx):
...
...         if ctx.category.cost < ctx.profile.balance:
...             return Success()
...         else:
...             return Failure()
...
...     async def persist_subscription(self, ctx):
...
...         ctx.subscription = await create_subscription(category=ctx.category, profile=ctx.profile)
...         return Success()
...
...     async def show_subscription(self, ctx):
...
...         return Result(ctx.subscription)

>>> asyncio.run(Subscribe().buy(category_id=1, profile_id=1))
Subscription(primary_key=9)

This code style allow you clearly separate actual business scenario from implementation details.

Note

stories library was heavily inspired by dry-transaction ruby gem.

— ⭐️ —

Drylabs maintains dry-python and helps those who want to use it inside their organizations.

Read more at drylabs.io