diff --git a/safe_transaction_service/tokens/migrations/0013_token_logo_uri.py b/safe_transaction_service/tokens/migrations/0013_token_logo_uri.py new file mode 100644 index 000000000..2523954c4 --- /dev/null +++ b/safe_transaction_service/tokens/migrations/0013_token_logo_uri.py @@ -0,0 +1,22 @@ +# Generated by Django 5.0.9 on 2024-10-23 14:34 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("tokens", "0012_alter_token_address_alter_token_copy_price"), + ] + + operations = [ + migrations.AddField( + model_name="token", + name="logo_uri", + field=models.URLField( + blank=True, + default="", + help_text="If provided, return this URI instead of the stored logo", + ), + ), + ] diff --git a/safe_transaction_service/tokens/models.py b/safe_transaction_service/tokens/models.py index 88b889fde..bf691a202 100644 --- a/safe_transaction_service/tokens/models.py +++ b/safe_transaction_service/tokens/models.py @@ -221,6 +221,11 @@ class Token(models.Model): format="PNG", processors=[Resize(256, 256, upscale=False)], ) + logo_uri = models.URLField( + default="", + blank=True, + help_text="If provided, return this URI instead of the stored logo", + ) events_bugged = models.BooleanField( default=False, help_text="Set `True` if token does not send `Transfer` event sometimes (e.g. WETH on minting)", @@ -278,21 +283,23 @@ def set_spam(self) -> None: def get_full_logo_uri(self) -> str: if self.logo: return self.logo.url - elif settings.AWS_S3_PUBLIC_URL: + + if self.logo_uri: + return self.logo_uri + + if settings.AWS_S3_PUBLIC_URL: return urljoin( settings.AWS_S3_PUBLIC_URL, get_token_logo_path( self, self.address + settings.TOKENS_LOGO_EXTENSION ), ) - else: - # Old behaviour - return urljoin( - settings.TOKENS_LOGO_BASE_URI, - get_token_logo_path( - self, self.address + settings.TOKENS_LOGO_EXTENSION - ), - ) + + # Old behaviour + return urljoin( + settings.TOKENS_LOGO_BASE_URI, + get_token_logo_path(self, self.address + settings.TOKENS_LOGO_EXTENSION), + ) def get_price_address(self) -> ChecksumAddress: """ diff --git a/safe_transaction_service/tokens/tasks.py b/safe_transaction_service/tokens/tasks.py index 87c28ca7e..7f3224eb2 100644 --- a/safe_transaction_service/tokens/tasks.py +++ b/safe_transaction_service/tokens/tasks.py @@ -7,6 +7,7 @@ from celery import app from celery.utils.log import get_task_logger +from eth_typing import ChecksumAddress from safe_eth.eth.ethereum_client import EthereumNetwork, get_auto_ethereum_client from safe_eth.eth.utils import fast_to_checksum_address from web3.exceptions import Web3Exception @@ -59,12 +60,30 @@ def fix_pool_tokens_task() -> Optional[int]: return number +def _parse_token_address_from_token_list( + token_address: str, +) -> Optional[ChecksumAddress]: + if token_address.startswith("0x"): # Ignore ENS names + return fast_to_checksum_address(token_address) + else: + # Try ENS resolve + ethereum_client = get_auto_ethereum_client() + try: + if resolved_address := ethereum_client.w3.ens.address(token_address): + return resolved_address + except (ValueError, Web3Exception): + logger.warning("Cannot resolve %s ENS address", token_address) + return None + + @app.shared_task() @close_gevent_db_connection_decorator def update_token_info_from_token_list_task() -> int: """ If there's at least one valid token list with at least 1 token, every token in the DB is marked as `not trusted` - and then every token on the list is marked as `trusted` + and then every token on the list is marked as `trusted`. + + `logoURI` is also stored for the tokens with logos :return: Number of tokens marked as `trusted` """ @@ -75,29 +94,21 @@ def update_token_info_from_token_list_task() -> int: except TokenListRetrievalException: logger.error("Cannot read tokens from %s", token_list) - if not tokens: - return 0 - - # Make sure current chainId matches the one in the list current_chain_id = get_ethereum_network().value - ethereum_client = get_auto_ethereum_client() - - token_addresses = [] - for token in tokens: - if token.get("chainId") == current_chain_id: - token_address = token["address"] - if token_address.startswith("0x"): - token_addresses.append(fast_to_checksum_address(token_address)) - else: - # Try ENS resolve - try: - if resolved_address := ethereum_client.w3.ens.address( - token_address - ): - token_addresses.append(resolved_address) - except (ValueError, Web3Exception): - logger.warning("Cannot resolve %s ENS address", token_address) + # Some lists are meant to be used for multiple chains + filtered_tokens = [ + token for token in tokens if token.get("chainId") == current_chain_id + ] + if not filtered_tokens: + return 0 + + tokens_updated_count = 0 with transaction.atomic(): Token.objects.update(trusted=False) - return Token.objects.filter(address__in=token_addresses).update(trusted=True) + for token in filtered_tokens: + token_address = _parse_token_address_from_token_list(token["address"]) + tokens_updated_count += Token.objects.filter(address=token_address).update( + logo_uri=token.get("logoURI") or "", trusted=True + ) + return tokens_updated_count diff --git a/safe_transaction_service/tokens/tests/test_models.py b/safe_transaction_service/tokens/tests/test_models.py index d4dd88110..573886140 100644 --- a/safe_transaction_service/tokens/tests/test_models.py +++ b/safe_transaction_service/tokens/tests/test_models.py @@ -61,6 +61,11 @@ def test_token_get_full_logo_uri(self): t.get_full_logo_uri(), f"{assets_url}/tokens/logos/{t.address}.png" ) + random_url = "http://random-url/logo.png" + t.logo_uri = random_url + t.save(update_fields=["logo_uri"]) + self.assertEqual(t.get_full_logo_uri(), random_url) + def test_token_trusted_spam_queryset(self): spam_tokens = [TokenFactory(spam=True), TokenFactory(spam=True)] not_spam_tokens = [ diff --git a/safe_transaction_service/tokens/tests/test_tasks.py b/safe_transaction_service/tokens/tests/test_tasks.py index 5dac0b82a..d3665e257 100644 --- a/safe_transaction_service/tokens/tests/test_tasks.py +++ b/safe_transaction_service/tokens/tests/test_tasks.py @@ -52,9 +52,18 @@ def test_update_token_info_from_token_list_task( self.assertEqual(update_token_info_from_token_list_task.delay().result, 0) # Create a token in the list, it should be updated - token = TokenFactory(address="0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2") + token = TokenFactory( + address="0x4A64515E5E1d1073e83f30cB97BEd20400b66E10", logo=None + ) self.assertFalse(token.trusted) self.assertEqual(update_token_info_from_token_list_task.delay().result, 1) + token.refresh_from_db() + self.assertTrue(token.trusted) + self.assertEqual( + token.logo_uri, + "https://cloudflare-ipfs.com/ipfs/QmYNLKHDEoG9FLJtbJ1r8HCyi7by9gksuacRkhkakxwEQ8", + ) + self.assertEqual(token.get_full_logo_uri(), token.logo_uri) # Create another token in the list, both should be updated token_2 = TokenFactory(address="0x2260FAC5E5542a773Aa44fBCfeDf7C193bc2C599")