Skip to content

http_storage

Base class for HTTP based stores.

API_PREFIX = settings.API_PREFIX module-attribute

API URL prefix.

HttpStorage

Bases: Storage

An HTTP base storage for a given resource type.

Source code in mlte/store/common/http_storage.py 
 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
class HttpStorage(Storage):
    """An HTTP base storage for a given resource type."""

    def __init__(
        self,
        uri: StoreURI,
        resource_type: ResourceType,
        client: Optional[OAuthHttpClient] = None,
    ) -> None:
        super().__init__(uri)

        if client is None:
            client = RequestsClient()
        self.client = client
        """The client for requests."""

        # Get credentials, if any, from the uri and into the client.
        uri.uri = self.client.process_credentials(uri.uri)
        self.clean_url = uri.uri
        """Store the clean URL without credentials."""

        self.resource_url = self.build_resource_url(resource_type.value)
        """Set the base URL for all calls to this resource."""

    def build_resource_url(self, resource: str) -> str:
        return f"{self.clean_url}{API_PREFIX}/{resource}"

    def start_session(self):
        """Perform authentication."""
        self.client.authenticate(f"{self.clean_url}{API_PREFIX}")

    def post(
        self, json: Any, groups: OrderedDict[str, str] = OrderedDict()
    ) -> Any:
        """Post method, to create resource."""
        return self.send_command(MethodType.POST, json=json, groups=groups)

    def put(
        self, json: Any, groups: OrderedDict[str, str] = OrderedDict()
    ) -> Any:
        """Put method, to update resource."""
        return self.send_command(MethodType.PUT, json=json, groups=groups)

    def get(
        self,
        id: Optional[str] = None,
        groups: OrderedDict[str, str] = OrderedDict(),
        query_args: dict[str, str] = {},
    ) -> Any:
        """Get method, to read resource."""
        return self.send_command(
            MethodType.GET, id=id, groups=groups, query_args=query_args
        )

    def delete(
        self, id: str, groups: OrderedDict[str, str] = OrderedDict()
    ) -> Any:
        """Delete method, to remove resource."""
        return self.send_command(MethodType.DELETE, id=id, groups=groups)

    def send_command(
        self,
        method: MethodType,
        groups: OrderedDict[str, str] = OrderedDict(),
        id: Optional[str] = None,
        query_args: dict[str, str] = {},
        json: Optional[Any] = None,
        resource_type: Optional[str] = None,
    ) -> Any:
        """
        Sends an HTTP command request to the backend API, and returns a JSON response from it. Commonly not used directly, as it is expected
        for callers to use the post(), put(), read() and delete() method, but can be used when a non-standard command, which does not properly
        match the arguments of the previous 4 methods, needs to be sent.

        :param method: what HTTP method to use, from MethodType
        :param groups: a dict of groups to prepend to the request path, where the key is the id of the given group, and the value is the string used to denote
        the beginning of the next nested level (e.g., {"model1": "version"}, with model1 being a model id, and "version" being the keyword for the next level).
        :param id: the id of a given resource.
        :param query_args: dictionary of query args to append.
        :param json: json payload to send, only for POST and PUT commands.
        :param resource_type: optional param to overwride the default resource type set up when creating this storage;
        string to be used as the first item in the URL's path.
        """
        path_url = ""

        # Add groups to path.
        for group_id, subgroup_name in groups.items():
            path_url += f"/{url_utils.make_valid_url_part(group_id)}/{url_utils.make_valid_url_part(subgroup_name)}"

        # Add id to path, if any.
        if id:
            path_url += f"/{url_utils.make_valid_url_part(id)}"

        # Add query args.
        query = ""
        link_char = "?"
        for arg_name, arg_value in query_args.items():
            query += f"{link_char}{url_utils.make_valid_url_part(arg_name)}={url_utils.make_valid_url_part(arg_value)}"
            link_char = "&"

        # Allow changing the resource part of the base resource URL, if needed.
        base_url = self.resource_url
        if resource_type:
            base_url = self.build_resource_url(resource_type)

        url = f"{base_url}{url_parse.quote(path_url)}{query}"
        if method == MethodType.POST:
            res = self.client.post(url, json=json)
        elif method == MethodType.PUT:
            res = self.client.put(url, json=json)
        elif method == MethodType.GET:
            res = self.client.get(url)
        elif method == MethodType.DELETE:
            res = self.client.delete(url)
        else:
            raise RuntimeError(f"Invalid method type: {method}")

        self.client.raise_for_response(res)
        return res.json()

clean_url = uri.uri instance-attribute

Store the clean URL without credentials.

client = client instance-attribute

The client for requests.

resource_url = self.build_resource_url(resource_type.value) instance-attribute

Set the base URL for all calls to this resource.

delete(id, groups=OrderedDict())

Delete method, to remove resource.

Source code in mlte/store/common/http_storage.py 
75
76
77
78
79
def delete(
    self, id: str, groups: OrderedDict[str, str] = OrderedDict()
) -> Any:
    """Delete method, to remove resource."""
    return self.send_command(MethodType.DELETE, id=id, groups=groups)

get(id=None, groups=OrderedDict(), query_args={})

Get method, to read resource.

Source code in mlte/store/common/http_storage.py 
64
65
66
67
68
69
70
71
72
73
def get(
    self,
    id: Optional[str] = None,
    groups: OrderedDict[str, str] = OrderedDict(),
    query_args: dict[str, str] = {},
) -> Any:
    """Get method, to read resource."""
    return self.send_command(
        MethodType.GET, id=id, groups=groups, query_args=query_args
    )

post(json, groups=OrderedDict())

Post method, to create resource.

Source code in mlte/store/common/http_storage.py 
52
53
54
55
56
def post(
    self, json: Any, groups: OrderedDict[str, str] = OrderedDict()
) -> Any:
    """Post method, to create resource."""
    return self.send_command(MethodType.POST, json=json, groups=groups)

put(json, groups=OrderedDict())

Put method, to update resource.

Source code in mlte/store/common/http_storage.py 
58
59
60
61
62
def put(
    self, json: Any, groups: OrderedDict[str, str] = OrderedDict()
) -> Any:
    """Put method, to update resource."""
    return self.send_command(MethodType.PUT, json=json, groups=groups)

send_command(method, groups=OrderedDict(), id=None, query_args={}, json=None, resource_type=None)

Sends an HTTP command request to the backend API, and returns a JSON response from it. Commonly not used directly, as it is expected for callers to use the post(), put(), read() and delete() method, but can be used when a non-standard command, which does not properly match the arguments of the previous 4 methods, needs to be sent.

Parameters:

Name Type Description Default
method MethodType

what HTTP method to use, from MethodType

 required
groups OrderedDict[str, str]

a dict of groups to prepend to the request path, where the key is the id of the given group, and the value is the string used to denote the beginning of the next nested level (e.g., {"model1": "version"}, with model1 being a model id, and "version" being the keyword for the next level).

 OrderedDict()
id Optional[str]

the id of a given resource.

 None
query_args dict[str, str]

dictionary of query args to append.

 {}
json Optional[Any]

json payload to send, only for POST and PUT commands.

 None
resource_type Optional[str]

optional param to overwride the default resource type set up when creating this storage; string to be used as the first item in the URL's path.

 None
Source code in mlte/store/common/http_storage.py 
 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
def send_command(
    self,
    method: MethodType,
    groups: OrderedDict[str, str] = OrderedDict(),
    id: Optional[str] = None,
    query_args: dict[str, str] = {},
    json: Optional[Any] = None,
    resource_type: Optional[str] = None,
) -> Any:
    """
    Sends an HTTP command request to the backend API, and returns a JSON response from it. Commonly not used directly, as it is expected
    for callers to use the post(), put(), read() and delete() method, but can be used when a non-standard command, which does not properly
    match the arguments of the previous 4 methods, needs to be sent.

    :param method: what HTTP method to use, from MethodType
    :param groups: a dict of groups to prepend to the request path, where the key is the id of the given group, and the value is the string used to denote
    the beginning of the next nested level (e.g., {"model1": "version"}, with model1 being a model id, and "version" being the keyword for the next level).
    :param id: the id of a given resource.
    :param query_args: dictionary of query args to append.
    :param json: json payload to send, only for POST and PUT commands.
    :param resource_type: optional param to overwride the default resource type set up when creating this storage;
    string to be used as the first item in the URL's path.
    """
    path_url = ""

    # Add groups to path.
    for group_id, subgroup_name in groups.items():
        path_url += f"/{url_utils.make_valid_url_part(group_id)}/{url_utils.make_valid_url_part(subgroup_name)}"

    # Add id to path, if any.
    if id:
        path_url += f"/{url_utils.make_valid_url_part(id)}"

    # Add query args.
    query = ""
    link_char = "?"
    for arg_name, arg_value in query_args.items():
        query += f"{link_char}{url_utils.make_valid_url_part(arg_name)}={url_utils.make_valid_url_part(arg_value)}"
        link_char = "&"

    # Allow changing the resource part of the base resource URL, if needed.
    base_url = self.resource_url
    if resource_type:
        base_url = self.build_resource_url(resource_type)

    url = f"{base_url}{url_parse.quote(path_url)}{query}"
    if method == MethodType.POST:
        res = self.client.post(url, json=json)
    elif method == MethodType.PUT:
        res = self.client.put(url, json=json)
    elif method == MethodType.GET:
        res = self.client.get(url)
    elif method == MethodType.DELETE:
        res = self.client.delete(url)
    else:
        raise RuntimeError(f"Invalid method type: {method}")

    self.client.raise_for_response(res)
    return res.json()

start_session()

Perform authentication.

Source code in mlte/store/common/http_storage.py 
48
49
50
def start_session(self):
    """Perform authentication."""
    self.client.authenticate(f"{self.clean_url}{API_PREFIX}")