def foo(a, b, *, c=1, d=2)
синтаксиса.
accepted_data = {
key: value for key, value in ctx.__dict__[
'_Context__ns'
].items()
}
@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
print(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()
Как правильно организовать код в таком случае? Заранее спасибо!
Привет. Сразу скажу что красивого решения я пока не нашёл.
Я обычно такое поведение выношу в отдельную story, т.е. Result
возвращаю на верхнем уровне вложенности.
@dataclass
class TokenUseCase:
@story
@arguments('user')
def obtain_user_token(I):
I.obtain_token
I.return_token
@story
@arguments('user')
def obtain_token(I):
I.require_user_active
I.create_token
Таким образом мы можем вызвать obtain_user_token
как самостоятельную сторю, а obtain_token
как дочернюю сторю для какой-то более большой.
Что-то вроде такого:
```
# step
def load_image(self, ctx):
result = io_call(ctx.get_image_file(image_id=ctx.image_id))
if not is_successful(result):
return Failure('image_cannot_be_loaded')
ctx.image = unwrap(result)
return Success()
```
С целью обозначить что будет выполнена io операция, которая которая может вызвать исключение, и чтобы можно было это исключение транслировать на зарегистрированную Failure.
mappers
нужен, поэтому не факт что враппер станет частью stories.
И в идеале наверное, чтобы был какой-то маппер, для для более гранулярного сопоставления исключений с failures.
А необработанные летели наверх или маппились с дефолтным failure. Имхо))
```
# step
def load_image(self, ctx):
wrapped = io_call(ctx.get_image_file(image_id=ctx.image_id))
if not is_successful(wrapped):
return Failure(wrapped.failure)
ctx.image = unwrap(wrapped)
return Success()
class RegisteredFailures(Enum):
image_cannot_be_loaded = auto()
image_not_found = auto()
permission_denied = auto()
def io_call(): -> FailureResult
pass
class ImageFailureResult(FailureResult):
@property
def failure(self):
failure = self._map(self._exception, RegisteredFailures.image_cannot_be_loaded)
return failure
```