Skip to content

Fetchers

MetadataFetcher

Fetcher class that makes network requests for metadata-related information.

Attributes:

Name Type Description
timeout int

request timeout in seconds.
max_retries int

maximum number of request retries.
sess Session

a requests Session object.
Source code in offchain/metadata/fetchers/metadata_fetcher.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
@FetcherRegistry.register
class MetadataFetcher(BaseFetcher):
    """Fetcher class that makes network requests for metadata-related information.

    Attributes:
        timeout (int): request timeout in seconds.
        max_retries (int): maximum number of request retries.
        sess (requests.Session): a requests Session object.
    """

    def __init__(
        self,
        timeout: int = 30,
        max_retries: int = 0,
        async_adapter_configs: Optional[list[AdapterConfig]] = None,
    ) -> None:
        self.timeout = timeout
        self.max_retries = max_retries
        self.sess = requests.Session()
        self.async_sess = httpx.AsyncClient()
        self.async_adapter_configs = async_adapter_configs

    def register_adapter(self, adapter: Adapter, url_prefix: str):  # type: ignore[no-untyped-def]  # noqa: E501
        """Register an adapter to a url prefix.

        Args:
            adapter (Adapter): an Adapter instance to register.
            url_prefix (str): the url prefix to which the adapter should be registered.
        """
        self.sess.mount(url_prefix, adapter)

    def set_max_retries(self, max_retries: int):  # type: ignore[no-untyped-def]
        """Setter function for max retries

        Args:
            max_retries (int): new maximum number of request retries.
        """
        self.max_retries = max_retries

    def set_timeout(self, timeout: int):  # type: ignore[no-untyped-def]
        """Setter function for timeout

        Args:
            timeout (int): new request timeout in seconds.
        """
        self.timeout = timeout

    def _head(self, uri: str):  # type: ignore[no-untyped-def]
        return self.sess.head(uri, timeout=self.timeout, allow_redirects=True)

    def _get(self, uri: str):  # type: ignore[no-untyped-def]
        return self.sess.get(uri, timeout=self.timeout, allow_redirects=True)

    async def _gen(self, uri: str) -> httpx.Response:
        from offchain.metadata.pipelines.metadata_pipeline import (
            DEFAULT_ADAPTER_CONFIGS,
        )

        configs = DEFAULT_ADAPTER_CONFIGS

        if self.async_adapter_configs:
            configs = self.async_adapter_configs

        for adapter_config in configs:
            if any(uri.startswith(prefix) for prefix in adapter_config.mount_prefixes):
                adapter = adapter_config.adapter_cls(
                    host_prefixes=adapter_config.host_prefixes, **adapter_config.kwargs
                )
                return await adapter.gen_send(
                    url=uri, timeout=self.timeout, sess=self.async_sess
                )
        return await self.async_sess.get(
            uri, timeout=self.timeout, follow_redirects=True
        )

    def fetch_mime_type_and_size(self, uri: str) -> Tuple[str, int]:
        """Fetch the mime type and size of the content at a given uri.

        Args:
            uri (str): uri from which to fetch content mime type and size.

        Returns:
            tuple[str, int]: mime type and size
        """
        try:
            res = self._head(uri)
            # For any error status, try a get
            if 300 <= res.status_code < 600:
                res = self._get(uri)
            res.raise_for_status()
            headers = res.headers
            size = headers.get("content-length", 0)
            content_type = headers.get("content-type") or headers.get("Content-Type")
            if content_type is not None:
                content_type, _ = cgi.parse_header(content_type)

            return content_type, size
        except Exception as e:
            logger.error(
                f"Failed to fetch content-type and size from uri {uri}. Error: {e}"
            )
            raise

    async def gen_fetch_mime_type_and_size(self, uri: str) -> Tuple[str, int]:
        """Fetch the mime type and size of the content at a given uri.

        Args:
            uri (str): uri from which to fetch content mime type and size.

        Returns:
            tuple[str, int]: mime type and size
        """
        try:
            # try skip head request
            # res = await self._gen_head(uri)
            # # For any error status, try a get
            # if 300 <= res.status_code < 600:
            res = await self._gen(uri)
            res.raise_for_status()
            headers = res.headers
            size = headers.get("content-length", 0)
            content_type = headers.get("content-type") or headers.get("Content-Type")
            if content_type is not None:
                content_type, _ = cgi.parse_header(content_type)

            return content_type, size
        except Exception as e:
            logger.error(
                f"Failed to fetch content-type and size from uri {uri}. Error: {e}"
            )
            raise

    def fetch_content(self, uri: str) -> Union[dict, str]:  # type: ignore[type-arg]
        """Fetch the content at a given uri

        Args:
            uri (str): uri from which to fetch content.

        Returns:
            Union[dict, str]: content fetched from uri
        """
        try:
            res = self._get(uri)
            res.raise_for_status()
            if res.text.startswith("{"):
                return res.json()  # type: ignore[no-any-return]
            else:
                return res.text  # type: ignore[no-any-return]

        except Exception as e:
            raise Exception(f"Don't know how to fetch metadata for {uri=}. {str(e)}")

    async def gen_fetch_content(self, uri: str) -> Union[dict, str]:  # type: ignore[type-arg]  # noqa: E501
        """Async fetch the content at a given uri

        Args:
            uri (str): uri from which to fetch content.

        Returns:
            Union[dict, str]: content fetched from uri
        """
        try:
            res = await self._gen(uri)
            res.raise_for_status()
            if res.text.startswith("{"):
                return res.json()  # type: ignore[no-any-return]
            else:
                return res.text

        except Exception as e:
            raise Exception(f"Don't know how to fetch metadata for {uri=}. {str(e)}")

fetch_content(uri)

Fetch the content at a given uri

Parameters:

Name Type Description Default
uri str

uri from which to fetch content.

 required

Returns:

Type Description
Union[dict, str]

Union[dict, str]: content fetched from uri
Source code in offchain/metadata/fetchers/metadata_fetcher.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def fetch_content(self, uri: str) -> Union[dict, str]:  # type: ignore[type-arg]
    """Fetch the content at a given uri

    Args:
        uri (str): uri from which to fetch content.

    Returns:
        Union[dict, str]: content fetched from uri
    """
    try:
        res = self._get(uri)
        res.raise_for_status()
        if res.text.startswith("{"):
            return res.json()  # type: ignore[no-any-return]
        else:
            return res.text  # type: ignore[no-any-return]

    except Exception as e:
        raise Exception(f"Don't know how to fetch metadata for {uri=}. {str(e)}")

fetch_mime_type_and_size(uri)

Fetch the mime type and size of the content at a given uri.

Parameters:

Name Type Description Default
uri str

uri from which to fetch content mime type and size.

 required

Returns:

Type Description
Tuple[str, int]

tuple[str, int]: mime type and size
Source code in offchain/metadata/fetchers/metadata_fetcher.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def fetch_mime_type_and_size(self, uri: str) -> Tuple[str, int]:
    """Fetch the mime type and size of the content at a given uri.

    Args:
        uri (str): uri from which to fetch content mime type and size.

    Returns:
        tuple[str, int]: mime type and size
    """
    try:
        res = self._head(uri)
        # For any error status, try a get
        if 300 <= res.status_code < 600:
            res = self._get(uri)
        res.raise_for_status()
        headers = res.headers
        size = headers.get("content-length", 0)
        content_type = headers.get("content-type") or headers.get("Content-Type")
        if content_type is not None:
            content_type, _ = cgi.parse_header(content_type)

        return content_type, size
    except Exception as e:
        logger.error(
            f"Failed to fetch content-type and size from uri {uri}. Error: {e}"
        )
        raise

gen_fetch_content(uri) async

Async fetch the content at a given uri

Parameters:

Name Type Description Default
uri str

uri from which to fetch content.

 required

Returns:

Type Description
Union[dict, str]

Union[dict, str]: content fetched from uri
Source code in offchain/metadata/fetchers/metadata_fetcher.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
async def gen_fetch_content(self, uri: str) -> Union[dict, str]:  # type: ignore[type-arg]  # noqa: E501
    """Async fetch the content at a given uri

    Args:
        uri (str): uri from which to fetch content.

    Returns:
        Union[dict, str]: content fetched from uri
    """
    try:
        res = await self._gen(uri)
        res.raise_for_status()
        if res.text.startswith("{"):
            return res.json()  # type: ignore[no-any-return]
        else:
            return res.text

    except Exception as e:
        raise Exception(f"Don't know how to fetch metadata for {uri=}. {str(e)}")

gen_fetch_mime_type_and_size(uri) async

Fetch the mime type and size of the content at a given uri.

Parameters:

Name Type Description Default
uri str

uri from which to fetch content mime type and size.

 required

Returns:

Type Description
Tuple[str, int]

tuple[str, int]: mime type and size
Source code in offchain/metadata/fetchers/metadata_fetcher.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
async def gen_fetch_mime_type_and_size(self, uri: str) -> Tuple[str, int]:
    """Fetch the mime type and size of the content at a given uri.

    Args:
        uri (str): uri from which to fetch content mime type and size.

    Returns:
        tuple[str, int]: mime type and size
    """
    try:
        # try skip head request
        # res = await self._gen_head(uri)
        # # For any error status, try a get
        # if 300 <= res.status_code < 600:
        res = await self._gen(uri)
        res.raise_for_status()
        headers = res.headers
        size = headers.get("content-length", 0)
        content_type = headers.get("content-type") or headers.get("Content-Type")
        if content_type is not None:
            content_type, _ = cgi.parse_header(content_type)

        return content_type, size
    except Exception as e:
        logger.error(
            f"Failed to fetch content-type and size from uri {uri}. Error: {e}"
        )
        raise

register_adapter(adapter, url_prefix)

Register an adapter to a url prefix.

Parameters:

Name Type Description Default
adapter Adapter

an Adapter instance to register.

 required
url_prefix str

the url prefix to which the adapter should be registered.

 required
Source code in offchain/metadata/fetchers/metadata_fetcher.py 
35
36
37
38
39
40
41
42
def register_adapter(self, adapter: Adapter, url_prefix: str):  # type: ignore[no-untyped-def]  # noqa: E501
    """Register an adapter to a url prefix.

    Args:
        adapter (Adapter): an Adapter instance to register.
        url_prefix (str): the url prefix to which the adapter should be registered.
    """
    self.sess.mount(url_prefix, adapter)

set_max_retries(max_retries)

Setter function for max retries

Parameters:

Name Type Description Default
max_retries int

new maximum number of request retries.

 required
Source code in offchain/metadata/fetchers/metadata_fetcher.py 
44
45
46
47
48
49
50
def set_max_retries(self, max_retries: int):  # type: ignore[no-untyped-def]
    """Setter function for max retries

    Args:
        max_retries (int): new maximum number of request retries.
    """
    self.max_retries = max_retries

set_timeout(timeout)

Setter function for timeout

Parameters:

Name Type Description Default
timeout int

new request timeout in seconds.

 required
Source code in offchain/metadata/fetchers/metadata_fetcher.py 
52
53
54
55
56
57
58
def set_timeout(self, timeout: int):  # type: ignore[no-untyped-def]
    """Setter function for timeout

    Args:
        timeout (int): new request timeout in seconds.
    """
    self.timeout = timeout