Skip
works on substories, not current step. More info here: https://stories.readthedocs.io/en/latest/execution.html#skip
I take a quick overview of the opus library.
It really looks like the dry-transaction library for the Ruby language.
As we discussed earlier, the closest alternative for Python will be a DSL based on class attributes.
I still trying to wrap my head around this idea.
I notice a few things we definitely should do in stories:
We already implement the possibility to skip steps, but I dislike :if
statement next to the step definition. I harm readability in my opinion.
The same goes to :with
keyword.
Also, I think retries is out of the scope of the stories
library and should be implemented inside story steps by the final user.
Thanks for the sharing!
HI! Thanks for giving stories
a try!
Failure Protocols
is a complete but unreleased feature.
As you can see the only left thing to do is documentation.
It's in its final form and will not change in the near future.
But to use it, you should install from the master.
I hope this example will shed some light on it.
from enum import Enum, auto
from stories import Failure, Result, Success, arguments, story
# Define parent story
class YoutubeDownload:
@story
@arguments("url", "directory")
def download(I):
I.fetch_file
I.decode_media
I.print_result
def decode_media(self, ctx):
try:
...
except MP4FormatError:
return Failure(YoutubeErrors.invalid_media)
else:
return Success(...)
def print_result(I):
return Result(...)
def __init__(self, fetch_file):
self.fetch_file = fetch_file
# Define parent story failure protocol.
@YoutubeDownload.download.failures
class YoutubeErrors(Enum):
invalid_media = auto()
# Define sub-story.
class FetchVideo:
@story
@arguments("url", "directory")
def fetch(I):
I.create_temp_dir
I.open_stream
I.download_locally
I.move_to_target_dir
def create_temp_file(self, ctx):
try:
...
except IOError:
return Failure(FetchErrors.no_space_left)
else:
return Success(...)
def open_stream(self, ctx):
try:
...
except HTTP401:
return Failure(FetchErrors.unauthorized)
else:
return Success(...)
def download_locally(self, ctx):
try:
...
except IOError:
return Failure(FetchErrors.no_space_left)
else:
return Success(...)
def move_to_target_dir(self, ctx):
try:
...
except IOError:
return Failure(FetchErrors.no_space_left)
else:
return Success(...)
# Define sub-story failure protocol.
@FetchVideo.fetch.failures
class FetchErrors(Enum):
no_space_left = auto()
unauthorized = auto()
# Instantiate parent and sub stories.
fetch_video = FetchVideo().fetch
youtube_download = YoutubeDownload(fetch_video).download
# Run the story.
#
# Failures were merged from the story hierarchy, so we can check them
# in one place. Use `youtube_download.failures` instead of
# `YoutubeErrors` or `FetchErrors`.
result = youtube_download.run(url="...", directory="...")
if result.is_success:
...
elif result.failed_because(youtube_download.failures.no_space_left):
...
elif result.failed_because(youtube_download.failures.unauthorized):
...
elif result.failed_because(youtube_download.failures.invalid_media):
...
Feel free to ask any questions about code snippet above.
Best regards,
Artem.
Skip
to the Break
dry-python/stories#141ctx
property of the story.run() result.
Hi,
at the moment my code look like this:
class BifPackager:
def __init__(self):
self.s3_storage = Storage()
self.cleanup_storage_path = CleanupStoragePathStep(self.s3_storage).call
@story
@arguments('params')
def package(I):
I.cleanup_storage_path
I.download_movie
I.generate_bif
I.upload_file
I.to_result
def download_movie(self, ctx):
return try_to_failure(
lambda: Success(downloaded_movie=download_to_temp(ctx.params['movie_path'])), 'download_error')
def generate_bif(self, ctx):
return try_to_failure(lambda: Success(bif_file=BifTool().process(ctx.downloaded_movie)), 'bif_error')
def upload_file(self, ctx):
return try_to_failure(
lambda: Success(
upload_file=self.s3_storage.upload_to_link(ctx.params['output_path'],
'{0}-hd.bif'.format(ctx.downloaded_movie))),
'upload_error')
def to_result(self, ctx):
return Result({'status': True})
failures_in(BifPackager, ['cleanup_storage_error', 'download_error', 'bif_error', 'upload_error', 'cleanup_error'])
and try_to_failuire
look like this:
def try_to_failure(fn, failure):
try:
return fn()
except Exception as e:
get_logger().error(e)
return Failure(failure)
Hi @alexandrubeu
Yes, we take inspiration from dry-transaction
but stories
isn't one to one clone of this library in Python.
We takes a little bit different approach. For example, we usually write story above in this way:
from dataclasses import dataclass
from stories import story, arguments, Success, Failure, Result
from stories.shortcuts import failures_in
@dataclass
class BifPackager:
@story
@arguments("params")
def package(I):
I.cleanup_storage_path
I.download_movie
I.generate_bif
I.upload_file
I.to_result
# Steps.
def download_movie(self, ctx):
downloaded_movie = download_to_temp(ctx.params["movie_path"])
if downloaded_movie:
return Success(downloaded_movie=downloaded_movie)
else:
return Failure("download_error")
def generate_bif(self, ctx):
bif_file = BifTool().process(ctx.downloaded_movie)
if bif_file:
return Success(bif_file=bif_file)
else:
return Failure("bif_error")
def upload_file(self, ctx):
uploaded_file = self.s3_storage.upload_to_link(
ctx.params["output_path"], "{0}-hd.bif".format(ctx.downloaded_movie)
)
if uploaded_file:
return Success(upload_file=uploaded_file)
else:
return Failure("upload_error")
def to_result(self, ctx):
return Result({"status": True})
# Dependencies.
s3_storage: Storage
cleanup_storage_path: story
failures_in(BifPackager, [
"cleanup_storage_error",
"download_error",
"bif_error",
"upload_error",
"cleanup_error",
])
storage = Storage()
cleanup = CleanupStoragePathStep(storage).call
package = BifPackager(s3_storage=storage, cleanup_storage_path=cleanup).package
result = package.run(params=movie_options)
if result.is_success:
return HTTPResponse(status=200)
elif result.failed_because('download_error'):
return HTTPResponse(
status=500,
data={
'error': 'download_error',
'message': f'Host response: {result.ctx.downloaded_movie}',
}
)
elif result.failed_because('cleanup_error'):
...
In the case of Failure
, the actual execution context is available in the result .
Why do we need this kind of separation?
Let's imagine the situation we want to use this story in a few places. For example, Django Web application build with forms, Django admin actions and REST framework API endpoint.
In each case interpretation of the same failure will be different.
In the future if web interface should support internationalization, Django admin and API endpoint should not know anything about it.
And if instead of building an actual response representation inside the story, we build it as close to the place where it actually used - we provide the necessary separation of concerns.
TL;DR; Build message outside of the story.
Hope that helps.
Regards,
Artem.
Привет!
stories
лучше всего использовать вместе с dependencies
.
Рассмотрим как пример код из tutorials. У нас есть:
Для DRF всё тоже готово. Можно взять, например, ModelViewSet. У него есть методы create, update, destroy так же как у form view есть form_valid.
Если есть желание, можешь запилить api слой для tutorials. Готов менторить.
Надеюсь было полезно.