Where communities thrive


  • Join over 1.5M+ people
  • Join over 100K+ communities
  • Free without limits
  • Create your own community
People
Repo info
Activity
    Artem Malyshev
    @proofit404
    @Morozzzko Skip works on substories, not current step. More info here: https://stories.readthedocs.io/en/latest/execution.html#skip
    As for wrapping each step with decorator/context manager: dry-python/stories#14
    Artem Malyshev
    @proofit404
    I just released stories 0.9! This is the first release which was announced in public :heart: https://stories.readthedocs.io/en/0.9/
    Ghost
    @ghost~576aa099c2f0db084a1f5559
    @proofit404 this new idea looks interesting. wanted to share with you https://github.com/zorbash/opus
    Artem Malyshev
    @proofit404
    Time learn
    elixir I guess))
    Artem Malyshev
    @proofit404

    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:

    • step metrics and instrumentation backends
    • cool pipeline pictures in the docs :)

    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!

    Yevhen Tsybulskyi
    @Hammer2900
    Hi, how use failure protocol ? When i send message to reason i get this error "Use 'failures' story method to define failure protocol." And in documentation i see only blank page.
    Artem Malyshev
    @proofit404

    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.

    Yevhen Tsybulskyi
    @Hammer2900
    @proofit404 thank you for examples ! now i do try it. I rewrite my old project with your DSL lib, when i finish do this i give you feedback =)
    Artem Malyshev
    @proofit404
    That's a great news! 🤘
    stories 0.10 is out! It comes with the brand new failure protocol feature.
    Serge Matveenko
    @lig
    @proofit404 good to know you are still on this. Unfortunately, I've been too busy lately to help.
    Artem Malyshev
    @proofit404
    Hi, @lig It's totally ok. We work on this from time to time.
    Artem Malyshev
    @proofit404
    The next stories release will have Flask debug toolbar support thanks to Max Hollmann!
    dry-python/stories#125
    Serge Matveenko
    @lig
    @proofit404
    Artem Malyshev
    @proofit404
    There is an interesting suggestion to rename Skip to the Break dry-python/stories#141
    I'm totally agree with this initiative, but maybe other have a better ideas.
    Alexandru Beu
    @alexandrubeu
    hey,
    Just started to use stories... And do have a problem related to Failure protocol. It seems that we only identify step and the type of Failure
    What if I want to decorate the Failure object with some metadata about the Failure?
    to have something like Failure(reson, data={}) ?
    Artem Malyshev
    @proofit404
    Hi, thanks for trying the project!
    It's not possible a the moment.
    But as a temporal workaround, you can access context ctx property of the story.run() result.
    Artem Malyshev
    @proofit404
    By the way, can you get us a concrete example, so we can decide a better API for that feature?
    Alexandru Beu
    @alexandrubeu

    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)
    Alexandru Beu
    @alexandrubeu
    What I think it would really help me is to have kwargs on Failure Protocol,
    something like Failure(protocol, exception=Exception, message=message).
    Alexandru Beu
    @alexandrubeu
    from my perspective I would create the API like this:
      packager = packager.package.run(movie_options)
            packager.do(success=lambda (result): do_stuff(result),
                        failure=lambda (failure): {
                            'error': failure.error,
                            'message': failure.message,
                            'protocol': failure.protocol
                        })
    On our Ruby projects we used for quite a while this library which is really great: https://dry-rb.org/gems/dry-transaction/
    Artem Malyshev
    @proofit404

    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",
    ])
    Artem Malyshev
    @proofit404
    So we will run this story and build a message for the actual API response in the code handling story result.
    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'):
        ...
    Artem Malyshev
    @proofit404

    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.

    Юлай Хасанов
    @yulai46_gitlab
    @proofit404 Здравствуйте! У меня вопрос о том как правильно использовать stories. Я пытаюсь понять как использовать его вместе с django rest framework. Например, где сериализовать модели? Внутри stories как один из шагов или я должен внутри сериализатора вызывать методы stories?
    Был бы рад если бы вы хотя бы направили меня в правильном направлении.
    Artem Malyshev
    @proofit404

    Привет!

    stories лучше всего использовать вместе с dependencies.

    Рассмотрим как пример код из tutorials. У нас есть:

    1. Классическая форма Django. Как и обычно занимается она только изъятием данных из запроса и валидацией типов полей. Никаких проверок связанных полей!
    2. Django Form View, собранная из Injector. В нем через this нотацию достаются уже валидные данные и передаются в собранную стори.
    3. Обёртка для story, в которой прописана реализация методов, которые нужны самой story.
    4. Класс story, который принимает в виде аргументов данные из формы.

    Для DRF всё тоже готово. Можно взять, например, ModelViewSet. У него есть методы create, update, destroy так же как у form view есть form_valid.

    Если есть желание, можешь запилить api слой для tutorials. Готов менторить.

    Надеюсь было полезно.

    Юлай Хасанов
    @yulai46_gitlab
    Спасибо за ответ! Желание запилить api есть. В том же репозитории еще один application добавить? Назвать его api например.
    Artem Malyshev
    @proofit404
    Привет! Я предлагаю создать форк и пустой PR, в котором продолжить обсуждение))
    Юлай Хасанов
    @yulai46_gitlab
    @proofit404 форк и PR создал: dry-python/tutorials#56
    Artem Malyshev
    @proofit404
    Всё увидел, спасибо. Сейчас опишу первые шаги.
    MuslimBeibytuly
    @MuslimBeibytuly
    Добрый вечер, есть ли где-либо примеры работы dry-python:stories+dependencies+returns с fastapi? будут ли проблемы с асинхронным фреймворком?
    Artem Malyshev
    @proofit404
    Привет. stories и returns скорее взаимоисключающие решения. Оба для сценариев. Dependencies не умеют собирать fastapi endpoint из контейнера, но можно обращаться к контейнеру напрямую из в view.
    Если есть желание реализовать поддержку fastapi для dependencies - готов менторить.
    Поддержка async def для stories есть в отдельном pr, но он ещё не смержен в мастер.
    MuslimBeibytuly
    @MuslimBeibytuly
    @proofit404 go, я только разбираюсь с тем, как вообще организовать ddd в проекте, но в fastapi без такой архитектуры нечего делать, быстро забьётся кучей непонятного кода
    Artem Malyshev
    @proofit404
    Можно посмотреть на https://github.com/dry-python/bookshelf
    Тут основной принцип 1 в 1 как в чистой архитектуре Мартина.
    В entities пишем бизнес правила (dataclasses без frameworkов), в usecases пишем бизнес сценарии (stories), реализацию в usecases прокидываем через dependencies.
    FastAPI views выглядят одной строкой, которая обращается к инжектору и передаёт аргументы из запроса в usecase.
    MuslimBeibytuly
    @MuslimBeibytuly
    @proofit404 не нашел документацию о том, как можно тестить отдельный шаг в stories, как воссоздать ctx для функции
    Artem Malyshev
    @proofit404
    Отдельные шаги в story тестировать не надо. Надо тестировать поведение стори целеком.
    Если есть необходимость тестирования отдельного шага - значит скорее всего в нём написана реализация завязанная на особенности системы, а не бизнес логика.
    В таком случае детали реализации лучше вынести в отдельную функцию, протестировать её отдельно от story, и закинуть в инстанс стори уже через dependencies.
    Например:
    1. Шаг в котором нет деталей реализации
    2. Отдельно детали реализации
    3. Добавляем одно в другое через конструктор
    MuslimBeibytuly
    @MuslimBeibytuly
    @proofit404 понял ошибку, можно ли вытащить все аргументы из ctx как dict? хочу засунуть все в kwargs
    Artem Malyshev
    @proofit404
    Это довольно хрупкое решение, которое делает интерфейс вызова неявным. Лучше передавать агрументы явно.