Skip to content

https

Download strategy class for http/https

HTTPDownloadContent

Bases: AttrDict

Class for returning values from Download HTTPS strategy.

Source code in oteapi/strategies/download/https.py 
110
111
112
113
class HTTPDownloadContent(AttrDict):
    """Class for returning values from Download HTTPS strategy."""

    key: str = Field(..., description="Key to access the data in the cache.")

key: str = Field(..., description='Key to access the data in the cache.') class-attribute instance-attribute

HTTPSConfig

Bases: AttrDict

HTTP(S)-specific Configuration Data Model.

Source code in oteapi/strategies/download/https.py 
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
class HTTPSConfig(AttrDict):
    """HTTP(S)-specific Configuration Data Model."""

    datacache_config: Optional[DataCacheConfig] = Field(
        None,
        description=(
            "Configurations for the data cache for storing the downloaded file "
            "content."
        ),
    )

    http_method: Literal["GET", "POST"] = Field(
        "GET",
        description=(
            "HTTP method to use for the download request. Only GET and POST are "
            "supported."
        ),
    )

    headers: Optional[dict[str, str]] = Field(
        None,
        description="HTTP headers to be included in the download request.",
    )

    cookies: Optional[dict[str, str]] = Field(
        None,
        description="Cookies to be included in the download request.",
    )

    query_parameters: Optional[dict[str, Union[str, list[str]]]] = Field(
        None,
        description=(
            "Query parameters to be included in the download request. Note, these can "
            "be included directly in the `downloadURL` as well."
        ),
    )

    post_body: Optional[Union[dict[str, Any], list[tuple[str, Any]], bytes]] = Field(
        None,
        description=(
            "The body of the POST request. This can be a a dictionary, list of tuples "
            "or bytes. This field is mutually exclusive with `post_body_json`."
        ),
    )

    post_body_json: Optional[Any] = Field(
        None,
        description=(
            "The body of the POST request as a JSON serializable Python object. This "
            "will be serialized to JSON and sent as the body of the POST request. "
            "This field is mutually exclusive with `post_body`."
        ),
    )

    @field_validator("http_method", mode="before")
    @classmethod
    def _upper_case_http_method(cls, value: Any) -> Any:
        if isinstance(value, str):
            return value.upper()
        return value

    @model_validator(mode="after")
    def _validate_post_bodies(self) -> HTTPSConfig:
        if self.http_method == "GET" and (self.post_body or self.post_body_json):
            warnings.warn(
                "POST body is provided for a GET requests - it will be ignored.",
                stacklevel=2,
            )
            self.post_body = None
            self.post_body_json = None
        if self.post_body and self.post_body_json:
            raise ValueError(
                "Only one of post_body and post_body_json can be provided."
            )
        return self

cookies: Optional[dict[str, str]] = Field(None, description='Cookies to be included in the download request.') class-attribute instance-attribute

datacache_config: Optional[DataCacheConfig] = Field(None, description='Configurations for the data cache for storing the downloaded file content.') class-attribute instance-attribute

headers: Optional[dict[str, str]] = Field(None, description='HTTP headers to be included in the download request.') class-attribute instance-attribute

http_method: Literal['GET', 'POST'] = Field('GET', description='HTTP method to use for the download request. Only GET and POST are supported.') class-attribute instance-attribute

post_body: Optional[Union[dict[str, Any], list[tuple[str, Any]], bytes]] = Field(None, description='The body of the POST request. This can be a a dictionary, list of tuples or bytes. This field is mutually exclusive with `post_body_json`.') class-attribute instance-attribute

post_body_json: Optional[Any] = Field(None, description='The body of the POST request as a JSON serializable Python object. This will be serialized to JSON and sent as the body of the POST request. This field is mutually exclusive with `post_body`.') class-attribute instance-attribute

query_parameters: Optional[dict[str, Union[str, list[str]]]] = Field(None, description='Query parameters to be included in the download request. Note, these can be included directly in the `downloadURL` as well.') class-attribute instance-attribute

HTTPSResourceConfig

Bases: ResourceConfig

HTTP(S) download strategy filter config.

Source code in oteapi/strategies/download/https.py 
 99
100
101
102
103
104
105
106
107
class HTTPSResourceConfig(ResourceConfig):
    """HTTP(S) download strategy filter config."""

    downloadUrl: AnyHttpUrl = Field(  # type: ignore[assignment]
        ..., description="The HTTP(S) URL, which will be downloaded."
    )
    configuration: HTTPSConfig = Field(
        HTTPSConfig(), description="HTTP(S) download strategy-specific configuration."
    )

configuration: HTTPSConfig = Field(HTTPSConfig(), description='HTTP(S) download strategy-specific configuration.') class-attribute instance-attribute

downloadUrl: AnyHttpUrl = Field(..., description='The HTTP(S) URL, which will be downloaded.') class-attribute instance-attribute

HTTPSStrategy

Strategy for retrieving data via http.

Registers strategies:

  • ("scheme", "http")
  • ("scheme", "https")
Source code in oteapi/strategies/download/https.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
144
145
146
147
148
149
150
151
152
153
154
@dataclass
class HTTPSStrategy:
    """Strategy for retrieving data via http.

    **Registers strategies**:

    - `("scheme", "http")`
    - `("scheme", "https")`

    """

    download_config: HTTPSResourceConfig

    def initialize(self) -> AttrDict:
        """Initialize."""
        return AttrDict()

    def get(self) -> HTTPDownloadContent:
        """Download via http/https and store on local cache."""
        cache = DataCache(self.download_config.configuration.datacache_config)
        if cache.config.accessKey and cache.config.accessKey in cache:
            key = cache.config.accessKey
        else:
            req = requests.request(
                method=self.download_config.configuration.http_method,
                url=str(self.download_config.downloadUrl),
                allow_redirects=True,
                timeout=(3, 27),  # timeout: (connect, read) in seconds
                headers=self.download_config.configuration.headers,
                cookies=self.download_config.configuration.cookies,
                params=self.download_config.configuration.query_parameters,
                # No reason to check the method is correct for sending content (POST),
                # since this is validated in the config model.
                data=self.download_config.configuration.post_body,
                json=self.download_config.configuration.post_body_json,
            )
            key = cache.add(req.content)

        return HTTPDownloadContent(key=key)

download_config: HTTPSResourceConfig instance-attribute

get()

Download via http/https and store on local cache.

Source code in oteapi/strategies/download/https.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def get(self) -> HTTPDownloadContent:
    """Download via http/https and store on local cache."""
    cache = DataCache(self.download_config.configuration.datacache_config)
    if cache.config.accessKey and cache.config.accessKey in cache:
        key = cache.config.accessKey
    else:
        req = requests.request(
            method=self.download_config.configuration.http_method,
            url=str(self.download_config.downloadUrl),
            allow_redirects=True,
            timeout=(3, 27),  # timeout: (connect, read) in seconds
            headers=self.download_config.configuration.headers,
            cookies=self.download_config.configuration.cookies,
            params=self.download_config.configuration.query_parameters,
            # No reason to check the method is correct for sending content (POST),
            # since this is validated in the config model.
            data=self.download_config.configuration.post_body,
            json=self.download_config.configuration.post_body_json,
        )
        key = cache.add(req.content)

    return HTTPDownloadContent(key=key)

initialize()

Initialize.

Source code in oteapi/strategies/download/https.py 
129
130
131
def initialize(self) -> AttrDict:
    """Initialize."""
    return AttrDict()