Skip to content

Adapters

IPFS

Provides an interface for Requests sessions to contact IPFS urls.

Parameters:

Name Type Description Default
host_prefixes list[str]

list of possible host url prefixes to choose from

 None
key str

optional key to send with request

 None
secret str

optional secret to send with request

 None
timeout int

request timeout in seconds. Defaults to 10 seconds.

 10
Source code in offchain/metadata/adapters/ipfs.py 
 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
@AdapterRegistry.register
class IPFSAdapter(HTTPAdapter):
    """Provides an interface for Requests sessions to contact IPFS urls.

    Args:
        host_prefixes (list[str], optional): list of possible host url prefixes to choose from
        key (str, optional): optional key to send with request
        secret (str, optional): optional secret to send with request
        timeout (int): request timeout in seconds. Defaults to 10 seconds.
    """  # noqa: E501

    def __init__(  # type: ignore[no-untyped-def]
        self,
        host_prefixes: Optional[list[str]] = None,
        key: Optional[str] = None,
        secret: Optional[str] = None,
        timeout: int = 10,
        *args,
        **kwargs,
    ):
        self.host_prefixes = host_prefixes or ["https://gateway.pinata.cloud/ipfs/"]

        assert all(
            [g.endswith("/") for g in self.host_prefixes]
        ), "gateways should have trailing slashes"

        self.key = key
        self.secret = secret
        self.timeout = timeout
        super().__init__(*args, **kwargs)

    def make_request_url(self, request_url: str, gateway: Optional[str] = None) -> str:
        """Parse and format incoming IPFS request url

        Args:
            request_url (str): incoming IPFS request url
            gateway (Optional[str]): gateway to use when making a request

        Returns:
            str: formatted IPFS url
        """

        gateway = gateway or random.choice(self.host_prefixes)
        return build_request_url(gateway=gateway, request_url=request_url)

    async def gen_send(self, url: str, sess: httpx.AsyncClient(), *args, **kwargs) -> httpx.Response:  # type: ignore[no-untyped-def, valid-type]  # noqa: E501
        """Format and send async request to IPFS host.

        Args:
            url (str): url to send request to
            sess (httpx.AsyncClient()): async client session

        Returns:
            httpx.Response: response from IPFS host.
        """
        return await sess.get(self.make_request_url(url), timeout=self.timeout, follow_redirects=True)  # type: ignore[no-any-return]  # noqa: E501

    def send(self, request: PreparedRequest, *args, **kwargs) -> Response:  # type: ignore[no-untyped-def]  # noqa: E501
        """For IPFS hashes, query pinata cloud gateway

        Args:
            request (PreparedRequest): incoming request

        Returns:
            Response: response from IPFS Gateway
        """
        request.url = self.make_request_url(request.url)  # type: ignore[arg-type]

        kwargs["timeout"] = self.timeout
        return super().send(request, *args, **kwargs)

gen_send(url, sess, *args, **kwargs) async

Format and send async request to IPFS host.

Parameters:

Name Type Description Default
url str

url to send request to

 required
sess AsyncClient

async client session

 required

Returns:

Type Description
Response

httpx.Response: response from IPFS host.
Source code in offchain/metadata/adapters/ipfs.py 
 99
100
101
102
103
104
105
106
107
108
109
async def gen_send(self, url: str, sess: httpx.AsyncClient(), *args, **kwargs) -> httpx.Response:  # type: ignore[no-untyped-def, valid-type]  # noqa: E501
    """Format and send async request to IPFS host.

    Args:
        url (str): url to send request to
        sess (httpx.AsyncClient()): async client session

    Returns:
        httpx.Response: response from IPFS host.
    """
    return await sess.get(self.make_request_url(url), timeout=self.timeout, follow_redirects=True)  # type: ignore[no-any-return]  # noqa: E501

make_request_url(request_url, gateway=None)

Parse and format incoming IPFS request url

Parameters:

Name Type Description Default
request_url str

incoming IPFS request url

 required
gateway Optional[str]

gateway to use when making a request

 None

Returns:

Name Type Description
str str

formatted IPFS url
Source code in offchain/metadata/adapters/ipfs.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
def make_request_url(self, request_url: str, gateway: Optional[str] = None) -> str:
    """Parse and format incoming IPFS request url

    Args:
        request_url (str): incoming IPFS request url
        gateway (Optional[str]): gateway to use when making a request

    Returns:
        str: formatted IPFS url
    """

    gateway = gateway or random.choice(self.host_prefixes)
    return build_request_url(gateway=gateway, request_url=request_url)

send(request, *args, **kwargs)

For IPFS hashes, query pinata cloud gateway

Parameters:

Name Type Description Default
request PreparedRequest

incoming request

 required

Returns:

Name Type Description
Response Response

response from IPFS Gateway
Source code in offchain/metadata/adapters/ipfs.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
def send(self, request: PreparedRequest, *args, **kwargs) -> Response:  # type: ignore[no-untyped-def]  # noqa: E501
    """For IPFS hashes, query pinata cloud gateway

    Args:
        request (PreparedRequest): incoming request

    Returns:
        Response: response from IPFS Gateway
    """
    request.url = self.make_request_url(request.url)  # type: ignore[arg-type]

    kwargs["timeout"] = self.timeout
    return super().send(request, *args, **kwargs)

ARWeave

Provides an interface for Requests sessions to contact ARWeave urls.

Parameters:

Name Type Description Default
host_prefixes list[str]

list of possible host url prefixes to choose from

 None
key str

optional key to send with request

 None
secret str

optional secret to send with request

 None
timeout int

request timeout in seconds. Defaults to 10 seconds.

 10
Source code in offchain/metadata/adapters/arweave.py 
12
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
@AdapterRegistry.register
class ARWeaveAdapter(HTTPAdapter):
    """Provides an interface for Requests sessions to contact ARWeave urls.

    Args:
        host_prefixes (list[str], optional): list of possible host url prefixes to choose from
        key (str, optional): optional key to send with request
        secret (str, optional): optional secret to send with request
        timeout (int): request timeout in seconds. Defaults to 10 seconds.
    """  # noqa: E501

    def __init__(  # type: ignore[no-untyped-def]
        self,
        host_prefixes: Optional[list[str]] = None,
        key: Optional[str] = None,
        secret: Optional[str] = None,
        timeout: int = 10,
        *args,
        **kwargs,
    ):
        self.host_prefixes = host_prefixes or ["https://arweave.net/"]

        assert all(
            [g.endswith("/") for g in self.host_prefixes]
        ), "gateways should have trailing slashes"

        self.key = key
        self.secret = secret
        self.timeout = timeout
        super().__init__(*args, **kwargs)

    def parse_ar_url(self, url: str) -> str:
        """Format and send async request to ARWeave host.

        Args:
            url (str): url to send request to
            sess (httpx.AsyncClient()): async client

        Returns:
            httpx.Response: response from ARWeave host.
        """
        parsed = parse_url(url)
        if parsed.scheme == "ar":
            gateway = random.choice(self.host_prefixes)
            new_url = f"{gateway}{parsed.host}"
            if parsed.path is not None:
                new_url += parsed.path
            url = new_url
        return url

    async def gen_send(self, url: str, sess: httpx.AsyncClient(), *args, **kwargs) -> httpx.Response:  # type: ignore[no-untyped-def, valid-type]  # noqa: E501
        """Format and send async request to ARWeave host.

        Args:
            url (str): url to send request to
            sess (httpx.AsyncClient()): async client

        Returns:
            httpx.Response: response from ARWeave host.
        """
        return await sess.get(self.parse_ar_url(url), timeout=self.timeout, follow_redirects=True)  # type: ignore[no-any-return]  # noqa: E501

    def send(self, request: PreparedRequest, *args, **kwargs) -> Response:  # type: ignore[no-untyped-def]  # noqa: E501
        """Format and send request to ARWeave host.

        Args:
            request (PreparedRequest): incoming request

        Returns:
            Response: response from ARWeave host.
        """
        request.url = self.parse_ar_url(request.url)  # type: ignore[arg-type]
        kwargs["timeout"] = self.timeout
        return super().send(request, *args, **kwargs)

gen_send(url, sess, *args, **kwargs) async

Format and send async request to ARWeave host.

Parameters:

Name Type Description Default
url str

url to send request to

 required
sess AsyncClient

async client

 required

Returns:

Type Description
Response

httpx.Response: response from ARWeave host.
Source code in offchain/metadata/adapters/arweave.py 
62
63
64
65
66
67
68
69
70
71
72
async def gen_send(self, url: str, sess: httpx.AsyncClient(), *args, **kwargs) -> httpx.Response:  # type: ignore[no-untyped-def, valid-type]  # noqa: E501
    """Format and send async request to ARWeave host.

    Args:
        url (str): url to send request to
        sess (httpx.AsyncClient()): async client

    Returns:
        httpx.Response: response from ARWeave host.
    """
    return await sess.get(self.parse_ar_url(url), timeout=self.timeout, follow_redirects=True)  # type: ignore[no-any-return]  # noqa: E501

parse_ar_url(url)

Format and send async request to ARWeave host.

Parameters:

Name Type Description Default
url str

url to send request to

 required
sess AsyncClient

async client

 required

Returns:

Type Description
str

httpx.Response: response from ARWeave host.
Source code in offchain/metadata/adapters/arweave.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def parse_ar_url(self, url: str) -> str:
    """Format and send async request to ARWeave host.

    Args:
        url (str): url to send request to
        sess (httpx.AsyncClient()): async client

    Returns:
        httpx.Response: response from ARWeave host.
    """
    parsed = parse_url(url)
    if parsed.scheme == "ar":
        gateway = random.choice(self.host_prefixes)
        new_url = f"{gateway}{parsed.host}"
        if parsed.path is not None:
            new_url += parsed.path
        url = new_url
    return url

send(request, *args, **kwargs)

Format and send request to ARWeave host.

Parameters:

Name Type Description Default
request PreparedRequest

incoming request

 required

Returns:

Name Type Description
Response Response

response from ARWeave host.
Source code in offchain/metadata/adapters/arweave.py 
74
75
76
77
78
79
80
81
82
83
84
85
def send(self, request: PreparedRequest, *args, **kwargs) -> Response:  # type: ignore[no-untyped-def]  # noqa: E501
    """Format and send request to ARWeave host.

    Args:
        request (PreparedRequest): incoming request

    Returns:
        Response: response from ARWeave host.
    """
    request.url = self.parse_ar_url(request.url)  # type: ignore[arg-type]
    kwargs["timeout"] = self.timeout
    return super().send(request, *args, **kwargs)

Data URI

Provides an interface for Requests sessions to handle data uris.

Source code in offchain/metadata/adapters/data_uri.py 
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
@AdapterRegistry.register
class DataURIAdapter(BaseAdapter):
    """Provides an interface for Requests sessions to handle data uris."""

    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def]
        super().__init__(*args, **kwargs)  # type: ignore[no-untyped-call]

    async def gen_send(self, url: str, *args, **kwargs) -> httpx.Response:  # type: ignore[no-untyped-def]  # noqa: E501
        """Handle async data uri request.

        Args:
            url (str): url

        Returns:
            httpx.Response: encoded data uri response.
        """
        response = httpx.Response(
            status_code=200,
            text=decode_data_url(url),  # type: ignore[no-untyped-call]
            request=httpx.Request(method="GET", url=url),
        )
        return response

    def send(self, request: PreparedRequest, *args, **kwargs):  # type: ignore[no-untyped-def]  # noqa: E501
        """Handle data uri request.

        Args:
            request (PreparedRequest): incoming request

        Returns:
            Response: encoded data uri response.
        """
        newResponse = Response()
        newResponse.request = request
        newResponse.url = request.url  # type: ignore[assignment]
        newResponse.connection = self  # type: ignore[attr-defined]
        try:
            response = urlopen(request.url)  # type: ignore[arg-type]
            newResponse.status_code = 200
            newResponse.headers = response.headers
            newResponse.raw = response
            newResponse.encoding = "utf-8"
            self.response = response
        finally:
            return newResponse

    def close(self):  # type: ignore[no-untyped-def]
        self.response.close()

gen_send(url, *args, **kwargs) async

Handle async data uri request.

Parameters:

Name Type Description Default
url str

url

 required

Returns:

Type Description
Response

httpx.Response: encoded data uri response.
Source code in offchain/metadata/adapters/data_uri.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
async def gen_send(self, url: str, *args, **kwargs) -> httpx.Response:  # type: ignore[no-untyped-def]  # noqa: E501
    """Handle async data uri request.

    Args:
        url (str): url

    Returns:
        httpx.Response: encoded data uri response.
    """
    response = httpx.Response(
        status_code=200,
        text=decode_data_url(url),  # type: ignore[no-untyped-call]
        request=httpx.Request(method="GET", url=url),
    )
    return response

send(request, *args, **kwargs)

Handle data uri request.

Parameters:

Name Type Description Default
request PreparedRequest

incoming request

 required

Returns:

Name Type Description
Response

encoded data uri response.
Source code in offchain/metadata/adapters/data_uri.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def send(self, request: PreparedRequest, *args, **kwargs):  # type: ignore[no-untyped-def]  # noqa: E501
    """Handle data uri request.

    Args:
        request (PreparedRequest): incoming request

    Returns:
        Response: encoded data uri response.
    """
    newResponse = Response()
    newResponse.request = request
    newResponse.url = request.url  # type: ignore[assignment]
    newResponse.connection = self  # type: ignore[attr-defined]
    try:
        response = urlopen(request.url)  # type: ignore[arg-type]
        newResponse.status_code = 200
        newResponse.headers = response.headers
        newResponse.raw = response
        newResponse.encoding = "utf-8"
        self.response = response
    finally:
        return newResponse