raise
, ; «» , , .def divide(first: float, second: float) -> float:
return first / second
float
. - : result = divide(1, 0)
print('x / y = ', result)
print
, 1 0 – , ZeroDivisionError
. , , .int
, str
, , , . raise something()
. , . .ZeroDivisionError
, .def divide(first: float, second: float) -> float:
try:
return first / second
except ZeroDivisionError:
return 0.0
None
? , , None
( ), , - .: — , .
ZeroDivisionError
. - .divide
.except
, . , , .goto
, .import requests
def fetch_user_profile(user_id: int) -> 'UserProfile':
"""Fetches UserProfile dict from foreign API."""
response = requests.get('/api/users/{0}'.format(user_id))
response.raise_for_status()
return response.json()
except Exception: pass
. . .None
. . if something is not None:
, TypeError
. .User
UserNotFound
MissingUser
. , AnonymousUser
Django, . , .@dry-python/return
. - , .from returns.result import Result, Success, Failure
def divide(first: float, second: float) -> Result[float, ZeroDivisionError]:
try:
return Success(first / second)
except ZeroDivisionError as exc:
return Failure(exc)
Success
Failure
. Result
. , , Result[float, ZeroDivisionError]
Success[float]
, Failure[ZeroDivisionError]
.Failure
: .1 + divide(1, 0)
# => mypy error: Unsupported operand types for + ("int" and "Result[float, ZeroDivisionError]")
Result
, . .from returns.result import Result, Success, Failure
def divide(first: float, second: float) -> Result[float, ZeroDivisionError]:
try:
return Success('Done')
# => error: incompatible type "str"; expected "float"
except ZeroDivisionError as exc:
return Failure(0)
# => error: incompatible type "int"; expected "ZeroDivisionError"
map
, ;bind
, .Success(4).bind(lambda number: Success(number / 2))
# => Success(2)
Success(4).map(lambda number: number + 1)
# => Success(5)
.bind
.map
c Failure
:Failure(4).bind(lambda number: Success(number / 2))
# => Failure(4)
Failure(4).map(lambda number: number / 2)
# => Failure(4)
Failure(4).rescue(lambda number: Success(number + 1))
# => Success(5)
Failure(4).fix(lambda number: number / 2)
# => Success(2)
.unwrap()
.value_or()
:Success(1).unwrap()
# => 1
Success(0).value_or(None)
# => 0
Failure(0).value_or(None)
# => None
Failure(1).unwrap()
# => Raises UnwrapFailedError()
.unwrap()
?Result
:from returns.result import Result, Success, Failure
class CreateAccountAndUser(object):
"""Creates new Account-User pair."""
# TODO: we need to create a pipeline of these methods somehow...
def _validate_user(
self, username: str, email: str,
) -> Result['UserSchema', str]:
"""Returns an UserSchema for valid input, otherwise a Failure."""
def _create_account(
self, user_schema: 'UserSchema',
) -> Result['Account', str]:
"""Creates an Account for valid UserSchema's. Or returns a Failure."""
def _create_user(
self, account: 'Account',
) -> Result['User', str]:
"""Create an User instance. If user already exists returns Failure."""
class CreateAccountAndUser(object):
"""Creates new Account-User pair."""
def __call__(self, username: str, email: str) -> Result['User', str]:
"""Can return a Success(user) or Failure(str_reason)."""
return self._validate_user(
username, email,
).bind(
self._create_account,
).bind(
self._create_user,
)
# ...
.unwrap()
. ? . ? @pipeline
:from result.functions import pipeline
class CreateAccountAndUser(object):
"""Creates new Account-User pair."""
@pipeline
def __call__(self, username: str, email: str) -> Result['User', str]:
"""Can return a Success(user) or Failure(str_reason)."""
user_schema = self._validate_user(username, email).unwrap()
account = self._create_account(user_schema).unwrap()
return self._create_user(account)
# ...
.unwrap()
@pipeline
: , - .unwrap()
Failure[str]
, @pipeline
Failure[str]
. .Result
. @safe, . , :from returns.functions import safe
@safe
def divide(first: float, second: float) -> float:
return first / second
# is the same as:
def divide(first: float, second: float) -> Result[float, ZeroDivisionError]:
try:
return Success(first / second)
except ZeroDivisionError as exc:
return Failure(exc)
@safe
, .@safe
. :import requests
from returns.functions import pipeline, safe
from returns.result import Result
class FetchUserProfile(object):
"""Single responsibility callable object that fetches user profile."""
#: You can later use dependency injection to replace `requests`
#: with any other http library (or even a custom service).
_http = requests
@pipeline
def __call__(self, user_id: int) -> Result['UserProfile', Exception]:
"""Fetches UserProfile dict from foreign API."""
response = self._make_request(user_id).unwrap()
return self._parse_json(response)
@safe
def _make_request(self, user_id: int) -> requests.Response:
response = self._http.get('/api/users/{0}'.format(user_id))
response.raise_for_status()
return response
@safe
def _parse_json(self, response: requests.Response) -> 'UserProfile':
return response.json()
@safe
, . Result[OldReturnType, Exception]
.Result
, ..unwrap()
, .@pipeline
, .unwrap
.Result
, ..fix()
.rescue()
.? Moscow Python Conf++ 5 , ! – dry-python core- Django Channels. dry-python -.
Source: https://habr.com/ru/post/445234/
All Articles