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. У нас есть:
- Классическая форма Django. Как и обычно занимается она только изъятием данных из запроса и валидацией типов полей. Никаких проверок связанных полей!
- Django Form View, собранная из Injector. В нем через this нотацию достаются уже валидные данные и передаются в собранную стори.
- Обёртка для story, в которой прописана реализация методов, которые нужны самой story.
- Класс story, который принимает в виде аргументов данные из формы.
Для DRF всё тоже готово. Можно взять, например, ModelViewSet. У него есть методы create, update, destroy так же как у form view есть form_valid.
Если есть желание, можешь запилить api слой для tutorials. Готов менторить.
Надеюсь было полезно.
Тут основной принцип 1 в 1 как в чистой архитектуре Мартина.
В entities пишем бизнес правила (dataclasses без frameworkов), в usecases пишем бизнес сценарии (stories), реализацию в usecases прокидываем через dependencies.
FastAPI views выглядят одной строкой, которая обращается к инжектору и передаёт аргументы из запроса в usecase.
Если есть необходимость тестирования отдельного шага - значит скорее всего в нём написана реализация завязанная на особенности системы, а не бизнес логика.
В таком случае детали реализации лучше вынести в отдельную функцию, протестировать её отдельно от story, и закинуть в инстанс стори уже через dependencies.
Например:
def foo(a, b, *, c=1, d=2) синтаксиса.
accepted_data = {
key: value for key, value in ctx.__dict__[
'_Context__ns'
].items()
}Знаю что костыль, но с другой стороны понимаю что лень все 20 аргументов отправлять в сервис руками:c есть ли какой-нибудь в arguments указывать не только название аргумента, но и описание в виде typing аннотации?(pydantic модель, это не по ddd, однако мы не стремимся получить идеальный ddd, лучше понятный по stories, но не слишком раздутый код)
Если шаг проверяет своё условие и понимет, что ему ничего делать ненужно - он возвращает Success. Обработка переходит к следующему шагу.
Eсли шаг понимает что условие удовлетворено - вызвает необходимую обработку, возвращает Skip. Обработки данный истории прекращается.
Eсли обратобка условия это что-то сложное, лучше оформить саму обработку в отдельную историю в отдельном классе.
А вот уже шаг проверки условия преобразовать во вложенную историю из двух шагов. Первый проверяет условию, второй - история из отдельного класса.
@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()).buyprint(ctx). Но в коде будет грязь.
@dataclass
class TokenUseCase:
@story
@arguments('user')
def obtain_token(I):
I.require_user_active
I.create_token
I.return_token
def require_user_active(self, ctx):
if ctx.user.is_active:
return Success()
return Failure()
def create_token(self, ctx):
ctx.token = self.create_user_token(ctx.user)
return Success()
def return_token(self, ctx):
return Result({'token': ctx.token})
# deps
create_user_token: Callable@proofit404 Добрый вечер! Например у меня есть TokenUseCase, который я хочу использовать напрямую для получения токена.
result = TokenUseCase(create_user_token=create_user_token).obtain_token.run(user=user)
token = result.value['token']Но также я хочу переиспользовать TokenUseCase.obtain_token как sub-story.
Но так сделать не получается, т.к. obtain_token возвращает Result и прекращает выполнение OtherUseCase.outer_story
obtain_token = TokenUseCase(create_user_token=create_user_token).obtain_token
OtherUseCase(find_token=obtain_token).outer_story.run()Как правильно организовать код в таком случае? Заранее спасибо!