Skip to content

Fix typing warnings #2534

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.

Sign up for GitHub

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
Uxio0 merged 1 commit into from
Jun 5, 2025
Merged

Fix typing warnings #2534

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
3 changes: 1 addition & 2 deletions safe_transaction_service/account_abstraction/helpers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import dataclasses
from typing import List

from eth_typing import ChecksumAddress
from safe_eth.eth import EthereumClient
Expand All @@ -19,7 +18,7 @@ class DecodedInitCode:
salt_nonce: int
expected_address: ChecksumAddress # Expected Safe deployment address
# Safe creation data
owners: List[ChecksumAddress]
owners: list[ChecksumAddress]
threshold: int
to: ChecksumAddress
data: bytes
Expand Down
12 changes: 6 additions & 6 deletions safe_transaction_service/account_abstraction/serializers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import datetime
from typing import Any, Dict, List, Optional
from typing import Any, Optional

from django.conf import settings
from django.db import transaction
Expand Down Expand Up @@ -39,9 +39,9 @@ class SafeOperationSignatureValidatorMixin:
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.ethereum_client = get_auto_ethereum_client()
self._deployment_owners: List[ChecksumAddress] = []
self._deployment_owners: list[ChecksumAddress] = []

def _get_owners(self, safe_address: ChecksumAddress) -> List[ChecksumAddress]:
def _get_owners(self, safe_address: ChecksumAddress) -> list[ChecksumAddress]:
"""
:param safe_address:
:return: `init_code` decoded owners if Safe is not deployed or current blockchain owners if Safe is deployed
Expand All @@ -59,7 +59,7 @@ def _validate_signature(
safe_operation_hash: bytes,
safe_operation_hash_preimage: bytes,
signature: bytes,
) -> List[SafeSignature]:
) -> list[SafeSignature]:
safe_owners = self._get_owners(safe_address)
parsed_signatures = SafeSignature.parse_signature(
signature,
Expand Down Expand Up @@ -375,7 +375,7 @@ def validate(self, attrs):
@transaction.atomic
def save(self, **kwargs):
safe_signatures = self.validated_data["safe_signatures"]
safe_operation_confirmations: List[SafeOperationConfirmation] = []
safe_operation_confirmations: list[SafeOperationConfirmation] = []
for safe_signature in safe_signatures:
safe_operation_confirmation, created = (
SafeOperationConfirmation.objects.get_or_create(
Expand Down Expand Up @@ -438,7 +438,7 @@ class SafeOperationResponseSerializer(serializers.Serializer):
confirmations = serializers.SerializerMethodField()
prepared_signature = serializers.SerializerMethodField()

def get_confirmations(self, obj: SafeOperationModel) -> Dict[str, Any]:
def get_confirmations(self, obj: SafeOperationModel) -> dict[str, Any]:
"""
Filters confirmations queryset

Expand Down
131 changes: 66 additions & 65 deletions safe_transaction_service/account_abstraction/services/aa_processor_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import logging
from functools import cache
from typing import List, Optional, Sequence, Tuple
from typing import Optional, Sequence

from django.conf import settings
from django.db import transaction
Expand Down Expand Up @@ -73,8 +73,8 @@ def __init__(
self.supported_entry_points = supported_entry_points

def get_user_operation_hashes_from_logs(
self, safe_address: ChecksumAddress, logs: [Sequence[LogReceipt]]
) -> List[HexBytes]:
self, safe_address: ChecksumAddress, logs: Sequence[LogReceipt]
) -> list[HexBytes]:
"""
:param safe_address:
:param logs:
Expand Down Expand Up @@ -109,7 +109,7 @@ def index_safe_operation_confirmations(
signature: bytes,
safe_operation_model: SafeOperationModel,
safe_operation: SafeOperation,
) -> List[SafeOperationConfirmationModel]:
) -> list[SafeOperationConfirmationModel]:
"""
Creates missing ``SafeOperationConfirmations``

Expand Down Expand Up @@ -144,7 +144,7 @@ def index_safe_operation(
user_operation_model: UserOperationModel,
user_operation: UserOperation,
user_operation_receipt: UserOperationReceipt,
) -> Optional[Tuple[SafeOperationModel, SafeOperation]]:
) -> Optional[tuple[SafeOperationModel, SafeOperation]]:
"""
Creates or updates a Safe Operation

Expand Down Expand Up @@ -206,7 +206,7 @@ def index_safe_operation(

def index_user_operation_receipt(
self, user_operation_model: UserOperationModel
) -> Tuple[UserOperationReceiptModel, UserOperationReceipt]:
) -> tuple[UserOperationReceiptModel, UserOperationReceipt]:
"""
Stores UserOperationReceipt. Can never be updated as if ``UserOperationReceipt`` is on database indexing
``UserOperation`` is not required
Expand Down Expand Up @@ -277,7 +277,7 @@ def index_user_operation(
safe_address: ChecksumAddress,
user_operation_hash: HexBytes,
ethereum_tx: history_models.EthereumTx,
) -> Tuple[UserOperationModel, UserOperation]:
) -> tuple[UserOperationModel, UserOperation] | None:
"""
Index ``UserOperation``, ``SafeOperation`` and ``UserOperationReceipt`` for the given ``UserOperation`` log

Expand All @@ -294,72 +294,73 @@ def index_user_operation(
safe_address,
user_operation_hash_hex,
)
else:
return None

logger.debug(
"[%s] Retrieving UserOperation from Bundler with user-operation-hash=%s on tx-hash=%s",
safe_address,
user_operation_hash_hex,
ethereum_tx.tx_hash,
)
user_operation = self.bundler_client.get_user_operation_by_hash(
user_operation_hash_hex
)
if not user_operation:
self.bundler_client.get_user_operation_by_hash.cache_clear()
raise BundlerClientException(
f"user-operation={user_operation_hash_hex} returned `null`"
)
if isinstance(user_operation, UserOperationV07):
raise UserOperationNotSupportedException(
f"user-operation={user_operation_hash_hex} for EntryPoint v0.7.0 is not supported"
)

try:
user_operation_model = UserOperationModel.objects.get(
hash=user_operation_hash_hex
)
logger.debug(
"[%s] Retrieving UserOperation from Bundler with user-operation-hash=%s on tx-hash=%s",
"[%s] Updating UserOperation with user-operation=%s on tx-hash=%s",
safe_address,
user_operation_hash_hex,
ethereum_tx.tx_hash,
)
user_operation = self.bundler_client.get_user_operation_by_hash(
user_operation_hash_hex
)
if not user_operation:
self.bundler_client.get_user_operation_by_hash.cache_clear()
raise BundlerClientException(
f"user-operation={user_operation_hash_hex} returned `null`"
)
if isinstance(user_operation, UserOperationV07):
raise UserOperationNotSupportedException(
f"user-operation={user_operation_hash_hex} for EntryPoint v0.7.0 is not supported"
)

try:
user_operation_model = UserOperationModel.objects.get(
hash=user_operation_hash_hex
)
logger.debug(
"[%s] Updating UserOperation with user-operation=%s on tx-hash=%s",
safe_address,
user_operation_hash_hex,
ethereum_tx.tx_hash,
)
user_operation_model.signature = user_operation.signature
user_operation_model.ethereum_tx = ethereum_tx
user_operation_model.save(update_fields=["signature", "ethereum_tx"])
except UserOperationModel.DoesNotExist:
logger.debug(
"[%s] Storing UserOperation with user-operation=%s on tx-hash=%s",
safe_address,
user_operation_hash_hex,
ethereum_tx.tx_hash,
)
user_operation_model = UserOperationModel.objects.create(
ethereum_tx=ethereum_tx,
hash=user_operation_hash_hex,
sender=user_operation.sender,
nonce=user_operation.nonce,
init_code=user_operation.init_code,
call_data=user_operation.call_data,
call_gas_limit=user_operation.call_gas_limit,
verification_gas_limit=user_operation.verification_gas_limit,
pre_verification_gas=user_operation.pre_verification_gas,
max_fee_per_gas=user_operation.max_fee_per_gas,
max_priority_fee_per_gas=user_operation.max_priority_fee_per_gas,
paymaster=user_operation.paymaster,
paymaster_data=user_operation.paymaster_data,
signature=user_operation.signature,
entry_point=user_operation.entry_point,
)

_, user_operation_receipt = self.index_user_operation_receipt(
user_operation_model
user_operation_model.signature = user_operation.signature
user_operation_model.ethereum_tx = ethereum_tx
user_operation_model.save(update_fields=["signature", "ethereum_tx"])
except UserOperationModel.DoesNotExist:
logger.debug(
"[%s] Storing UserOperation with user-operation=%s on tx-hash=%s",
safe_address,
user_operation_hash_hex,
ethereum_tx.tx_hash,
)
self.index_safe_operation(
user_operation_model, user_operation, user_operation_receipt
user_operation_model = UserOperationModel.objects.create(
ethereum_tx=ethereum_tx,
hash=user_operation_hash_hex,
sender=user_operation.sender,
nonce=user_operation.nonce,
init_code=user_operation.init_code,
call_data=user_operation.call_data,
call_gas_limit=user_operation.call_gas_limit,
verification_gas_limit=user_operation.verification_gas_limit,
pre_verification_gas=user_operation.pre_verification_gas,
max_fee_per_gas=user_operation.max_fee_per_gas,
max_priority_fee_per_gas=user_operation.max_priority_fee_per_gas,
paymaster=user_operation.paymaster,
paymaster_data=user_operation.paymaster_data,
signature=user_operation.signature,
entry_point=user_operation.entry_point,
)

return user_operation_model, user_operation
_, user_operation_receipt = self.index_user_operation_receipt(
user_operation_model
)
self.index_safe_operation(
user_operation_model, user_operation, user_operation_receipt
)

return user_operation_model, user_operation

def process_aa_transaction(
self, safe_address: ChecksumAddress, ethereum_tx: history_models.EthereumTx
Expand Down
3 changes: 1 addition & 2 deletions safe_transaction_service/analytics/services/analytics_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import json
from functools import cache
from typing import List

from safe_transaction_service.utils.redis import get_redis

Expand All @@ -13,7 +12,7 @@ def get_analytics_service() -> "AnalyticsService":
class AnalyticsService:
REDIS_TRANSACTIONS_PER_SAFE_APP = "analytics_transactions_per_safe_app"

def get_safe_transactions_per_safe_app(self) -> List:
def get_safe_transactions_per_safe_app(self) -> list[dict]:
redis = get_redis()
analytic_result = redis.get(self.REDIS_TRANSACTIONS_PER_SAFE_APP)
if analytic_result:
Expand Down
15 changes: 7 additions & 8 deletions safe_transaction_service/contracts/management/commands/setup_safe_contracts.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import logging
from typing import List, Tuple

from django.core.files import File
from django.core.management import BaseCommand, CommandError
Expand Down Expand Up @@ -108,16 +107,16 @@ def handle(self, *args, **options):

@staticmethod
def _get_deployments_by_chain_and_version(
versions: List[str], chain_id: str
) -> List[Tuple[str, str, str]]:
versions: list[str], chain_id: str
) -> list[tuple[str, str, str]]:
"""
Get the list of contracts for the given versions and chain.

:param versions: list of versions
:param chain_id: chain id
:return: list of (version, contract_name, contract_address)
"""
chain_deployments: List[Tuple[str, str, str]] = []
chain_deployments: list[tuple[str, str, str]] = []
for version in versions:
for contract_name, addresses in safe_deployments[version].items():
for contract_address in addresses.get(chain_id, []):
Expand All @@ -127,16 +126,16 @@ def _get_deployments_by_chain_and_version(

@staticmethod
def _get_default_deployments_by_version_on_chain(
versions: List[str], ethereum_client: EthereumClient
) -> List[Tuple[str, str, str]]:
versions: list[str], ethereum_client: EthereumClient
) -> list[tuple[str, str, str]]:
"""
Get the default deployments by version actually deployed on chain.

:param versions: list of versions
:param ethereum_client: Ethereum client
:return: list of (version, contract_name, contract_address)
"""
chain_deployments: List[Tuple[str, str, str]] = []
chain_deployments: list[tuple[str, str, str]] = []
for version in versions:
for contract_name, addresses in default_safe_deployments[version].items():
for contract_address in addresses:
Expand All @@ -149,7 +148,7 @@ def _get_default_deployments_by_version_on_chain(

@staticmethod
def _create_or_update_contracts_from_deployments(
deployments: List[Tuple[str, str, str]],
deployments: list[tuple[str, str, str]],
queryset,
force_update_contracts: bool,
logo_file: File,
Expand Down
6 changes: 3 additions & 3 deletions safe_transaction_service/contracts/models.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import operator
import os
from logging import getLogger
from typing import Any, Dict, List, Optional
from typing import Any, Optional

from django.conf import settings
from django.core.exceptions import ValidationError
Expand Down Expand Up @@ -38,7 +38,7 @@ def get_file_storage():
return default_storage


def validate_abi(value: Dict[str, Any]):
def validate_abi(value: dict[str, Any]):
try:
if not value:
raise ValueError("Empty ABI not allowed")
Expand Down Expand Up @@ -66,7 +66,7 @@ class ContractAbi(models.Model):
def __str__(self):
return f"ContractABI {self.relevance} - {self.description}"

def abi_functions(self) -> List[str]:
def abi_functions(self) -> list[str]:
return [x["name"] for x in self.abi if x["type"] == "function"]

def save(self, *args, **kwargs) -> None:
Expand Down
Loading