Skip to content

http

Implementation of HTTP user store

HttpGroupMapper

Bases: GroupMapper

HTTP mapper for the group resource.

Source code in mlte/store/user/underlying/http.py 
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
class HttpGroupMapper(GroupMapper):
    """HTTP mapper for the group resource."""

    def __init__(self, storage: HttpResourceStorage) -> None:
        self.storage = storage
        """The HTTP storage access."""

    def create(self, new_group: Group, context: Any = None) -> Group:
        response = self.storage.post(json=new_group.to_json())
        return Group(**response)

    def read(self, group_name: str, context: Any = None) -> Group:
        response = self.storage.get(id=group_name)
        return Group(**response)

    def list(self, context: Any = None) -> list[str]:
        response = self.storage.get()
        return typing.cast(list[str], response)

    def edit(self, group: Group, context: Any = None) -> Group:
        response = self.storage.put(json=group.to_json())
        return Group(**response)

    def delete(self, group_name: str, context: Any = None) -> Group:
        response = self.storage.delete(id=group_name)
        return Group(**response)

storage = storage instance-attribute

The HTTP storage access.

HttpPermissionMapper

Bases: PermissionMapper

HTTP mapper for the permission resource.

Source code in mlte/store/user/underlying/http.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class HttpPermissionMapper(PermissionMapper):
    """HTTP mapper for the permission resource."""

    def __init__(self, storage: HttpResourceStorage) -> None:
        self.storage = storage
        """The HTTP storage access."""

    def list(self, context: Any = None) -> list[str]:
        response = self.storage.get()
        return typing.cast(list[str], response)

    def list_details(
        self,
        context: Any = None,
        limit: int = ResourceMapper.DEFAULT_LIST_LIMIT,
        offset: int = 0,
    ) -> List[Any]:
        """
        Read details of resources within limit and offset.
        :param context: Any additional context needed for this resource.
        :param limit: The limit on resources to read
        :param offset: The offset on resources to read
        :return: The read resources
        """
        response = self.storage.get("s/permissions/details")
        return [permission for permission in response][offset : offset + limit]

    def read(self, permission_str: str, context: Any = None) -> Permission:
        response = self.storage.get(id=f"s/permission/{permission_str}")
        return Permission(**response)

storage = storage instance-attribute

The HTTP storage access.

list_details(context=None, limit=ResourceMapper.DEFAULT_LIST_LIMIT, offset=0)

Read details of resources within limit and offset.

Parameters:

Name Type Description Default
context Any

Any additional context needed for this resource.

 None
limit int

The limit on resources to read

 DEFAULT_LIST_LIMIT
offset int

The offset on resources to read

 0

Returns:

Type Description
List[Any]

The read resources
Source code in mlte/store/user/underlying/http.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def list_details(
    self,
    context: Any = None,
    limit: int = ResourceMapper.DEFAULT_LIST_LIMIT,
    offset: int = 0,
) -> List[Any]:
    """
    Read details of resources within limit and offset.
    :param context: Any additional context needed for this resource.
    :param limit: The limit on resources to read
    :param offset: The offset on resources to read
    :return: The read resources
    """
    response = self.storage.get("s/permissions/details")
    return [permission for permission in response][offset : offset + limit]

HttpUserMapper

Bases: UserMapper

HTTP mapper for the user resource.

Source code in mlte/store/user/underlying/http.py 
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
class HttpUserMapper(UserMapper):
    """HTTP mapper for the user resource."""

    def __init__(self, storage: HttpResourceStorage) -> None:
        self.storage = storage
        """The HTTP storage access."""

    def create(self, new_user: UserWithPassword, context: Any = None) -> User:
        response = self.storage.post(json=new_user.to_json())
        return User(**response)

    def read(self, entry_name: str, context: Any = None) -> User:
        response = self.storage.get(id=entry_name)
        return User(**response)

    def list(self, context: Any = None) -> list[str]:
        response = self.storage.get()
        return typing.cast(list[str], response)

    def edit(
        self, user: Union[UserWithPassword, BasicUser], context: Any = None
    ) -> User:
        response = self.storage.put(json=user.to_json())
        return User(**response)

    def delete(self, user_name: str, context: Any = None) -> User:
        response = self.storage.delete(id=user_name)
        return User(**response)

storage = storage instance-attribute

The HTTP storage access.

HttpUserStore

Bases: UserStore

A http implementation of the MLTE user store.

Source code in mlte/store/user/underlying/http.py 
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
class HttpUserStore(UserStore):
    """A http implementation of the MLTE user store."""

    def __init__(
        self, *, uri: StoreURI, client: Optional[OAuthHttpClient] = None
    ) -> None:
        self.user_storage = HttpResourceStorage(
            uri=uri, resource_type=ResourceType.USER, client=client
        )
        """HTTP user storage."""

        self.group_storage = HttpResourceStorage(
            uri=uri, resource_type=ResourceType.GROUP, client=client
        )
        """HTTP group storage."""

        self.permission_storage = HttpResourceStorage(
            uri=uri, resource_type="groups/permissions", client=client
        )
        """HTTP group storage."""

        # Adding default data is not done for remote stores, since remote ones already did it when they were started.
        super().__init__(uri=uri, add_default_data=False)

    def session(self) -> UserStoreSession:
        """
        Return a session handle for the store instance.
        :return: The session handle
        """
        return HttpUserStoreSession(
            user_storage=self.user_storage,
            group_storage=self.group_storage,
            permission_storage=self.permission_storage,
        )

group_storage = HttpResourceStorage(uri=uri, resource_type=(ResourceType.GROUP), client=client) instance-attribute

HTTP group storage.

permission_storage = HttpResourceStorage(uri=uri, resource_type='groups/permissions', client=client) instance-attribute

HTTP group storage.

user_storage = HttpResourceStorage(uri=uri, resource_type=(ResourceType.USER), client=client) instance-attribute

HTTP user storage.

session()

Return a session handle for the store instance.

Returns:

Type Description
UserStoreSession

The session handle
Source code in mlte/store/user/underlying/http.py 
50
51
52
53
54
55
56
57
58
59
def session(self) -> UserStoreSession:
    """
    Return a session handle for the store instance.
    :return: The session handle
    """
    return HttpUserStoreSession(
        user_storage=self.user_storage,
        group_storage=self.group_storage,
        permission_storage=self.permission_storage,
    )

HttpUserStoreSession

Bases: UserStoreSession

An HTTP implementation of the MLTE user store session.

Source code in mlte/store/user/underlying/http.py 
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
class HttpUserStoreSession(UserStoreSession):
    """An HTTP implementation of the MLTE user store session."""

    def __init__(
        self,
        *,
        user_storage: HttpResourceStorage,
        group_storage: HttpResourceStorage,
        permission_storage: HttpResourceStorage,
    ) -> None:
        self.user_storage = user_storage
        """HTTP user storage."""

        self.group_storage = group_storage
        """HTTP group storage."""

        self.user_mapper = HttpUserMapper(user_storage)
        """User mapper."""

        self.group_mapper = HttpGroupMapper(group_storage)
        """Group mapper."""

        self.permission_mapper = HttpPermissionMapper(permission_storage)
        """Group mapper."""

    def close(self):
        # No closing needed
        pass

group_mapper = HttpGroupMapper(group_storage) instance-attribute

Group mapper.

group_storage = group_storage instance-attribute

HTTP group storage.

permission_mapper = HttpPermissionMapper(permission_storage) instance-attribute

Group mapper.

user_mapper = HttpUserMapper(user_storage) instance-attribute

User mapper.

user_storage = user_storage instance-attribute

HTTP user storage.