dry-python/stories

Artem Malyshev

@proofit404

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. У нас есть:

Классическая форма Django. Как и обычно занимается она только изъятием данных из запроса и валидацией типов полей. Никаких проверок связанных полей!
Django Form View, собранная из Injector. В нем через this нотацию достаются уже валидные данные и передаются в собранную стори.
Обёртка для story, в которой прописана реализация методов, которые нужны самой story.
Класс 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.
Например:

MuslimBeibytuly

@MuslimBeibytuly

@proofit404 понял ошибку, можно ли вытащить все аргументы из ctx как dict? хочу засунуть все в kwargs

Artem Malyshev

@proofit404

Это довольно хрупкое решение, которое делает интерфейс вызова неявным. Лучше передавать агрументы явно.

Понимаю что лень писать, но это экономит время на чтение кода в дальнейшем. Можно, например, отделить keyword only arguments с помощью def foo(a, b, *, c=1, d=2) синтаксиса.

MuslimBeibytuly

@MuslimBeibytuly

@proofit404 в mr получил:

accepted_data = {
            key: value for key, value in ctx.__dict__[
                '_Context__ns'
            ].items()
        }

Знаю что костыль, но с другой стороны понимаю что лень все 20 аргументов отправлять в сервис руками:c есть ли какой-нибудь в arguments указывать не только название аргумента, но и описание в виде typing аннотации?(pydantic модель, это не по ddd, однако мы не стремимся получить идеальный ddd, лучше понятный по stories, но не слишком раздутый код)

Artem Malyshev

@proofit404

Если поишлёшь примеры кода, который является проблемным, я подумаю как это сделать лаконично.

Yevhen Ts.

@Hammer2900

я написал в личку, в общий чат будет немного не то скидывать код

Artem Malyshev

@proofit404

@MuslimBeibytuly your approach will not work from stories v0.11.2, pin your requirements or rewrite this hack.

I hope it isn't across all over your codebase :)

paveloder

@paveloder

Возможно ли использовать type hinting для контекста при наличии контракта, например:

class Story:
    @story
    def do(I):

        I.one

    def one(self, ctx: Context):

        return Success(bar=ctx.foo + "2")

@Story.do.contract
class Context(BaseModel):

    foo: str
    bar: str

как это правильно реализовать?

Artem Malyshev

@proofit404

Привет. Да, type hinting можно использовать для аннотаций контекста.

Есть пара идей для плагина для mypy чтобы автоматизировать некоторые шаблонные куски кода. Но в целом должно работать.

Artem Malyshev

@proofit404

После того, как смержим dry-python/stories#271 должно быть ещё лучше.

ramidk

@ramidk

@proofit404 Добрый день! А почему нет возможности получить код ошибки из Failure()? Например: result.failure_reason

Artem Malyshev

@proofit404

@ramidk А для каких целей он вам нужен? Если по ключу из dict достать или использовать в result.reason is Enun.foo, то это не получится. Для каждой композиции story собирается новый Enum.

Artem Malyshev

@proofit404

Потому что композиция может состоять из разных story.

Yevhen Ts.

@Hammer2900

вопрос такой, бывают такие случаи логики где применяются много if логики и она в таких случаях бывает заковыристая, как правильно оформлять стори в этом случае ?

вот маленький пример

нужно сделать много стори и соединить их ? или же можно как то переключать раут в стори на другой шаг ?

Artem Malyshev

@proofit404

Лучше сделать из этого одну плоскую историю. Каждый шаг в порядке приоритетности проверки условий.
Если шаг проверяет своё условие и понимет, что ему ничего делать ненужно - он возвращает Success. Обработка переходит к следующему шагу.
Eсли шаг понимает что условие удовлетворено - вызвает необходимую обработку, возвращает Skip. Обработки данный истории прекращается.
Eсли обратобка условия это что-то сложное, лучше оформить саму обработку в отдельную историю в отдельном классе.
А вот уже шаг проверки условия преобразовать во вложенную историю из двух шагов. Первый проверяет условию, второй - история из отдельного класса.

Yevhen Ts.

@Hammer2900

хорошо, спасибо, попробую, надеюсь это будет выглядеть красиво. Конечно бы пример в документации не помешал бы по этому поводу...

Artem Malyshev

@proofit404

Уже в процессе :)

Artem Malyshev

@proofit404

https://twitter.com/dry_py/status/1244370504955330563

ramidk

@ramidk

@proofit404 Добрый день! возможно ли (чтобы убрать лишний step вызова реализации) вместо:

>>> def find_price(self, ctx):
...     ctx.price = self.impl.find_price(ctx.price_id)
...     return Success()

>>> def __init__(self, impl):
...     self.impl = impl

Использовать:

>>> class Price:
...
...     @story
...     @arguments("category", "price_id")
...     def find_price(I):
...
...         I...
...         I...


>>> class Subscription(MethodDefinitions):
...
...     @story
...     @arguments("category_id", "price_id", "profile_id")
...     def buy(I):
...
...         I.find_category
...         I.impl.find_price
...
...        def __init__(self, impl):
...            self.impl = impl

Subscription(impl=Price()).buy

Where communities thrive

People

Repo info

Activity