Skip to content

Add black to web3's public utils #2513

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Jun 15, 2022
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions newsfragments/2513.misc.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Add black checks to ``web3/utils`` directory
2 changes: 1 addition & 1 deletion tox.ini
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ basepython=python
extras=linter
commands=
flake8 {toxinidir}/web3 {toxinidir}/ens {toxinidir}/ethpm {toxinidir}/tests --exclude {toxinidir}/ethpm/ethpm-spec
black {toxinidir}/ethpm --exclude {toxinidir}/ethpm/ethpm-spec --check
black {toxinidir}/ethpm {toxinidir}/web3/utils --exclude {toxinidir}/ethpm/ethpm-spec --check
isort --recursive --check-only --diff {toxinidir}/web3/ {toxinidir}/ens/ {toxinidir}/ethpm/ {toxinidir}/tests/
mypy -p web3 -p ethpm -p ens --config-file {toxinidir}/mypy.ini

Expand Down
63 changes: 33 additions & 30 deletions web3/utils/async_exception_handling.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,61 +32,64 @@ async def async_handle_offchain_lookup(
offchain_lookup_payload: Dict[str, Any],
transaction: TxParams,
) -> bytes:
formatted_sender = to_hex_if_bytes(offchain_lookup_payload['sender']).lower()
formatted_data = to_hex_if_bytes(offchain_lookup_payload['callData']).lower()
formatted_sender = to_hex_if_bytes(offchain_lookup_payload["sender"]).lower()
formatted_data = to_hex_if_bytes(offchain_lookup_payload["callData"]).lower()

if formatted_sender != to_hex_if_bytes(transaction['to']).lower():
if formatted_sender != to_hex_if_bytes(transaction["to"]).lower():
raise ValidationError(
'Cannot handle OffchainLookup raised inside nested call. Returned `sender` value does '
'not equal `to` address in transaction.'
"Cannot handle OffchainLookup raised inside nested call. Returned `sender` value does "
"not equal `to` address in transaction."
)

for url in offchain_lookup_payload['urls']:
for url in offchain_lookup_payload["urls"]:
formatted_url = URI(
str(url)
.replace('{sender}', str(formatted_sender))
.replace('{data}', str(formatted_data))
.replace("{sender}", str(formatted_sender))
.replace("{data}", str(formatted_data))
)

try:
if '{data}' in url and '{sender}' in url:
if "{data}" in url and "{sender}" in url:
response = await async_get_response_from_get_request(formatted_url)
elif '{sender}' in url:
response = await async_get_response_from_post_request(formatted_url, data={
"data": formatted_data,
"sender": formatted_sender
})
elif "{sender}" in url:
response = await async_get_response_from_post_request(
formatted_url,
data={"data": formatted_data, "sender": formatted_sender},
)
else:
raise ValidationError('url not formatted properly.')
raise ValidationError("url not formatted properly.")
except Exception:
continue # try next url if timeout or issues making the request

if 400 <= response.status <= 499: # if request returns 400 error, raise exception
if (
400 <= response.status <= 499
): # if request returns 400 error, raise exception
response.raise_for_status()
if not 200 <= response.status <= 299: # if not 400 error, try next url
continue

result = await async_get_json_from_client_response(response)

if 'data' not in result.keys():
if "data" not in result.keys():
raise ValidationError(
"Improperly formatted response for offchain lookup HTTP request - missing 'data' "
"field."
)

encoded_data_with_function_selector = b''.join([
# 4-byte callback function selector
to_bytes_if_hex(offchain_lookup_payload['callbackFunction']),

# encode the `data` from the result and the `extraData` as bytes
encode_abi(
['bytes', 'bytes'],
[
to_bytes_if_hex(result['data']),
to_bytes_if_hex(offchain_lookup_payload['extraData']),
]
)
])
encoded_data_with_function_selector = b"".join(
[
# 4-byte callback function selector
to_bytes_if_hex(offchain_lookup_payload["callbackFunction"]),
# encode the `data` from the result and the `extraData` as bytes
encode_abi(
["bytes", "bytes"],
[
to_bytes_if_hex(result["data"]),
to_bytes_if_hex(offchain_lookup_payload["extraData"]),
],
),
]
)

return encoded_data_with_function_selector
raise MultipleFailedRequests("Offchain lookup failed for supplied urls.")
66 changes: 36 additions & 30 deletions web3/utils/exception_handling.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,61 +31,67 @@ def handle_offchain_lookup(
offchain_lookup_payload: Dict[str, Any],
transaction: TxParams,
) -> bytes:
formatted_sender = to_hex_if_bytes(offchain_lookup_payload['sender']).lower()
formatted_data = to_hex_if_bytes(offchain_lookup_payload['callData']).lower()
formatted_sender = to_hex_if_bytes(offchain_lookup_payload["sender"]).lower()
formatted_data = to_hex_if_bytes(offchain_lookup_payload["callData"]).lower()

if formatted_sender != to_hex_if_bytes(transaction['to']).lower():
if formatted_sender != to_hex_if_bytes(transaction["to"]).lower():
raise ValidationError(
'Cannot handle OffchainLookup raised inside nested call. Returned `sender` value does '
'not equal `to` address in transaction.'
"Cannot handle OffchainLookup raised inside nested call. Returned `sender` value does "
"not equal `to` address in transaction."
)

for url in offchain_lookup_payload['urls']:
for url in offchain_lookup_payload["urls"]:
formatted_url = URI(
str(url)
.replace('{sender}', str(formatted_sender))
.replace('{data}', str(formatted_data))
.replace("{sender}", str(formatted_sender))
.replace("{data}", str(formatted_data))
)

try:
if '{data}' in url and '{sender}' in url:
if "{data}" in url and "{sender}" in url:
response = get_response_from_get_request(formatted_url)
elif '{sender}' in url:
response = get_response_from_post_request(formatted_url, data={
"data": formatted_data,
"sender": formatted_sender,
})
elif "{sender}" in url:
response = get_response_from_post_request(
formatted_url,
data={
"data": formatted_data,
"sender": formatted_sender,
},
)
else:
raise ValidationError('url not formatted properly.')
raise ValidationError("url not formatted properly.")
except Exception:
continue # try next url if timeout or issues making the request

if 400 <= response.status_code <= 499: # if request returns 400 error, raise exception
if (
400 <= response.status_code <= 499
): # if request returns 400 error, raise exception
response.raise_for_status()
if not 200 <= response.status_code <= 299: # if not 400 error, try next url
continue

result = response.json()

if 'data' not in result.keys():
if "data" not in result.keys():
raise ValidationError(
"Improperly formatted response for offchain lookup HTTP request - missing 'data' "
"field."
)

encoded_data_with_function_selector = b''.join([
# 4-byte callback function selector
to_bytes_if_hex(offchain_lookup_payload['callbackFunction']),

# encode the `data` from the result and the `extraData` as bytes
encode_abi(
['bytes', 'bytes'],
[
to_bytes_if_hex(result['data']),
to_bytes_if_hex(offchain_lookup_payload['extraData']),
]
)
])
encoded_data_with_function_selector = b"".join(
[
# 4-byte callback function selector
to_bytes_if_hex(offchain_lookup_payload["callbackFunction"]),
# encode the `data` from the result and the `extraData` as bytes
encode_abi(
["bytes", "bytes"],
[
to_bytes_if_hex(result["data"]),
to_bytes_if_hex(offchain_lookup_payload["extraData"]),
],
),
]
)

return encoded_data_with_function_selector
raise MultipleFailedRequests("Offchain lookup failed for supplied urls.")