Skip to content

Commit ebcc993

Browse files
author
Ivica Matic
committed
feat: adding more functions
1 parent ae0a8ba commit ebcc993

File tree

1 file changed

+291
-11
lines changed

1 file changed

+291
-11
lines changed

‎apim_subscriptions_manager/apim_subscriptions_manager.py‎

Lines changed: 291 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,22 @@ class APIMUserCreationError(Exception):
1818
pass
1919

2020

21+
class APIMUserNotFoundError(Exception):
22+
pass
23+
24+
25+
class APIMSubscriptionAlreadyExistsError(Exception):
26+
pass
27+
28+
29+
class APIMSubscriptionCreationError(Exception):
30+
pass
31+
32+
33+
class APIMSubscriptionNotFoundError(Exception):
34+
pass
35+
36+
2137
class ApimSubscriptionsManager:
2238
_tenant_id: str = None
2339
_client_id: str = None
@@ -84,16 +100,61 @@ def _get_api_token(self):
84100
self._api_token_expiry = token_expires_on
85101
return self._api_token
86102

103+
def _get_auth_headers(self) -> Dict[str, str]:
104+
"""
105+
Return auth headers required for interacting with APIM
106+
107+
:return: Auth headers
108+
"""
109+
headers = {
110+
"Authorization": f"Bearer {self._get_api_token()}",
111+
"Content-Type": "application/json",
112+
}
113+
return headers
114+
87115
def create_user_on_apim(self, user_id: str, email: str, first_name: str, last_name: str,
88116
group_name: str = None) -> Dict[str, Any]:
89117
"""
118+
Create a user on APIM
90119
91-
:param user_id:
92-
:param email:
93-
:param first_name:
94-
:param last_name:
95-
:param group_name:
96-
:return:
120+
:param user_id: Unique identifier for the user
121+
:param email: Email address of the user
122+
:param first_name: First name of the user
123+
:param last_name: Last name of the user
124+
:param group_name: Name of the group to which the user should be added
125+
:return: Details of the user created on APIM
126+
127+
Examples:
128+
>>> print(create_user_on_apim("123-unique-id-for-subscription-123", "test.user@spatialdays.com", "test", "user"))
129+
{
130+
"id":"<redacted>",
131+
"type":"Microsoft.ApiManagement/service/users",
132+
"name":"123-unique-id-for-subscription-123",
133+
"properties":{
134+
"firstName":"test",
135+
"lastName":"user",
136+
"email":"test.user@spatialdays.com",
137+
"state":"active",
138+
"registrationDate":"2023-06-01T09:33:15.997Z",
139+
"note":"None",
140+
"groups":[
141+
{
142+
"id":"<redacted>",
143+
"name":"Developers",
144+
"description":"Developers is a built-in group. Its membership is managed by the system. Signed-in users fall into this group.",
145+
"builtIn":true,
146+
"type":"system",
147+
"externalId":"None"
148+
}
149+
],
150+
"identities":[
151+
{
152+
"provider":"Basic",
153+
"id":"test.user@spatialdays.com"
154+
}
155+
]
156+
}
157+
}
97158
"""
98159

99160
if not user_id:
@@ -109,10 +170,7 @@ def create_user_on_apim(self, user_id: str, email: str, first_name: str, last_na
109170
f"{self._apim_rg_name}/providers/Microsoft.ApiManagement/service/{self._apim_name}/users/{user_id}"
110171
f"?api-version=2022-08-01")
111172

112-
headers = {
113-
"Authorization": f"Bearer {self._get_api_token()}",
114-
"Content-Type": "application/json",
115-
}
173+
headers = self._get_auth_headers()
116174

117175
body = json.dumps({
118176
"properties": {
@@ -127,7 +185,8 @@ def create_user_on_apim(self, user_id: str, email: str, first_name: str, last_na
127185
if response.status_code == 200:
128186
logging.error(f"User with id {user_id} already exists")
129187
raise APIMUserAlreadyExistsError(
130-
f"User with id {user_id} already exists. Status code: {response.status_code}, Response: {response.text}")
188+
f"User with id {user_id} already exists. Status code: {response.status_code}, \
189+
Response: {response.text}")
131190
elif response.status_code == 201:
132191
logging.debug(f"User with id {user_id} created successfully")
133192
if group_name:
@@ -148,3 +207,224 @@ def create_user_on_apim(self, user_id: str, email: str, first_name: str, last_na
148207
else:
149208
raise APIMUserCreationError(
150209
f"Failed to create user. Status code: {response.status_code}, Response: {response.text}")
210+
211+
def get_user_from_apim(self, user_id: str) -> Dict[str, Any]:
212+
"""
213+
Gets the user details from APIM
214+
215+
:param user_id: The unique identifier for the user
216+
:return: Details of the user from APIM
217+
218+
Examples:
219+
>>> print(get_user_from_apim("123-unique-id-for-subscription-123"))
220+
{
221+
'id': '/subscriptions/<redacted>/resourceGroups/<redacted>/providers/Microsoft.ApiManagement/service/<redacted>/users/123-unique-id-for-subscription-123',
222+
'type': 'Microsoft.ApiManagement/service/users',
223+
'name': '123-unique-id-for-subscription-123',
224+
'properties': {
225+
'firstName': 'test',
226+
'lastName': 'user',
227+
'email': 'test.user@spatialdays.com',
228+
'state': 'active',
229+
'registrationDate': '2023-06-01T09:33:15.997Z',
230+
'note': None,
231+
'identities': [{
232+
'provider': 'Basic',
233+
'id': 'test.user@spatialdays.com'
234+
}]
235+
}
236+
}
237+
"""
238+
url = (f"https://management.azure.com/subscriptions/{self._apim_subscription_id}/resourceGroups/"
239+
f"{self._apim_rg_name}/providers/Microsoft.ApiManagement/service/{self._apim_name}/users/"
240+
f"{user_id}?api-version=2022-08-01")
241+
242+
headers = self._get_auth_headers()
243+
244+
response = requests.get(url, headers=headers)
245+
246+
if response.status_code == 200:
247+
return response.json()
248+
else:
249+
raise APIMUserNotFoundError(
250+
f"User with id {user_id} not found. Status code: {response.status_code}, Response: {response.text}")
251+
252+
def delete_user_from_apim(self, user_id: str) -> str:
253+
"""
254+
Deletes a user from APIM
255+
256+
:param user_id: The unique identifier of the user to be deleted
257+
:return: The unique identifier of the user deleted
258+
Examples:
259+
>>> print(delete_user_from_apim("123-unique-id-for-subscription-123"))
260+
123-unique-id-for-subscription-123
261+
"""
262+
263+
url = (f"https://management.azure.com/subscriptions/{self._apim_subscription_id}/resourceGroups/"
264+
f"{self._apim_rg_name}/providers/Microsoft.ApiManagement/service/{self._apim_name}/users/{user_id}"
265+
f"?api-version=2022-08-01")
266+
267+
headers = self._get_auth_headers()
268+
269+
response = requests.delete(url, headers=headers)
270+
271+
if response.status_code == 200:
272+
return user_id
273+
elif response.status_code == 204:
274+
raise APIMUserNotFoundError(
275+
f"User with id {user_id} not found or couldn't be deleted. Status code: {response.status_code}, Response: {response.text}")
276+
else:
277+
raise APIMUserNotFoundError(
278+
f"User with id {user_id} not found or couldn't be deleted. Status code: {response.status_code}, Response: {response.text}")
279+
280+
def make_subscription_for_user_on_all_apis(self, user_id: str) -> Dict[str, Any]:
281+
"""
282+
Makes a subscription for a user on all APIs
283+
284+
:param user_id: The unique identifier of the user
285+
:return: Details of the subscription created
286+
287+
Examples:
288+
>>> print(make_subscription_for_user_on_all_apis("123-unique-id-for-subscription-123"))
289+
{
290+
'id': '/subscriptions/<redacted>/resourceGroups/<redacted>/providers/Microsoft.ApiManagement/service/<redacted>/subscriptions/123-unique-id-for-subscription-123',
291+
'type': 'Microsoft.ApiManagement/service/subscriptions',
292+
'name': '123-unique-id-for-subscription-123',
293+
'properties': {
294+
'ownerId': '/subscriptions/<redacted>/resourceGroups/<redacted>/providers/Microsoft.ApiManagement/service/<redacted>/users/123-unique-id-for-subscription-123',
295+
'user': {
296+
'id': '/subscriptions/<redacted>/resourceGroups/<redacted>/providers/Microsoft.ApiManagement/service/<redacted>/users/123-unique-id-for-subscription-123',
297+
'firstName': 'test',
298+
'lastName': 'user',
299+
'email': 'test.user@spatialdays.com',
300+
'state': 'active',
301+
'registrationDate': '2023-06-01T09:33:15.997Z',
302+
'note': None,
303+
'groups': [],
304+
'identities': []
305+
},
306+
'scope': '/subscriptions/<redacted>/resourceGroups/<redacted>/providers/Microsoft.ApiManagement/service/<redacted>/apis',
307+
'displayName': '123-unique-id-for-subscription-123',
308+
'state': 'active',
309+
'createdDate': '2023-06-01T09:37:51.0432183Z',
310+
'startDate': '2023-06-01T00:00:00Z',
311+
'expirationDate': None,
312+
'endDate': None,
313+
'notificationDate': None,
314+
'primaryKey': '<redacted>',
315+
'secondaryKey': '<redacted>',
316+
'stateComment': None,
317+
'allowTracing': False
318+
}
319+
}
320+
"""
321+
url = (f"https://management.azure.com/subscriptions/{self._apim_subscription_id}/resourceGroups/"
322+
f"{self._apim_rg_name}/providers/Microsoft.ApiManagement/service/{self._apim_name}"
323+
f"/subscriptions/{user_id}"
324+
f"?api-version=2022-08-01")
325+
326+
headers = {
327+
"Authorization": f"Bearer {self._get_api_token()}",
328+
"Content-Type": "application/json",
329+
}
330+
331+
body = json.dumps({
332+
"properties": {
333+
"scope": "/apis",
334+
"displayName": user_id,
335+
"state": "active",
336+
"ownerId": f"/users/{user_id}",
337+
}
338+
})
339+
340+
response = requests.put(url, headers=headers, data=body)
341+
342+
if response.status_code == 201:
343+
return response.json()
344+
elif response.status_code == 200:
345+
raise APIMSubscriptionAlreadyExistsError(
346+
f"Subscription for user with id {user_id} already exists. Status code: {response.status_code},"
347+
f" Response: {response.text}")
348+
else:
349+
raise APIMSubscriptionCreationError(
350+
f"Failed to create subscription. Status code: {response.status_code}, Response: {response.text}")
351+
352+
def get_subscription_for_user(self, user_id: str) -> Dict[str, Any]:
353+
"""
354+
Retrieves a subscription for a user from the API Management (APIM) service.
355+
356+
:param user_id: The unique identifier of the user for whom the subscription is to be retrieved.
357+
:return: dictionary containing the details of the user's subscription. This includes the subscription ID, type,
358+
name, associated properties such as owner ID, scope, display name, state, creation date, start date, expiration date,
359+
end date, notification date, state comment and whether tracing is allowed.
360+
Examples:
361+
>>> print(get_subscription_for_user("123-unique-id-for-subscription-123"))
362+
{
363+
'id': '/subscriptions/<redacted>/resourceGroups/<redacted>/providers/Microsoft.ApiManagement/service/<redacted>/subscriptions/123-unique-id-for-subscription-123',
364+
'type': 'Microsoft.ApiManagement/service/subscriptions',
365+
'name': '123-unique-id-for-subscription-123',
366+
'properties': {
367+
'ownerId': '/subscriptions/<redacted>/resourceGroups/<redacted>/providers/Microsoft.ApiManagement/service/<redacted>/users/123-unique-id-for-subscription-123',
368+
'scope': '/subscriptions/<redacted>/resourceGroups/<redacted>/providers/Microsoft.ApiManagement/service/<redacted>/apis',
369+
'displayName': '123-unique-id-for-subscription-123',
370+
'state': 'active',
371+
'createdDate': '2023-06-01T09:37:51.043Z',
372+
'startDate': '2023-06-01T00:00:00Z',
373+
'expirationDate': None,
374+
'endDate': None,
375+
'notificationDate': None,
376+
'stateComment': None,
377+
'allowTracing': False
378+
}
379+
}
380+
"""
381+
382+
url = (f"https://management.azure.com/subscriptions/{self._apim_subscription_id}/resourceGroups/"
383+
f"{self._apim_rg_name}/providers/Microsoft.ApiManagement/service/{self._apim_name}"
384+
f"/subscriptions/{user_id}"
385+
f"?api-version=2022-08-01")
386+
387+
headers = self._get_auth_headers()
388+
389+
response = requests.get(url, headers=headers)
390+
391+
if response.status_code == 200:
392+
return response.json()
393+
else:
394+
raise APIMSubscriptionNotFoundError(
395+
f"Subscription for user with id {user_id} not found. Status code: {response.status_code},"
396+
f" Response: {response.text}")
397+
398+
def get_subscription_secrets_for_user(self, user_id: str) -> Dict[str, str]:
399+
"""
400+
Retrieves the primary and secondary keys for a user's subscription from the API Management (APIM) service.
401+
402+
Args:
403+
user_id (str): The unique identifier of the user for whom the subscription keys are to be retrieved.
404+
405+
Returns:
406+
dict: A dictionary containing the primary and secondary keys for the user's subscription.
407+
408+
Examples:
409+
>>> print(get_subscription_secrets_for_user("123-unique-id-for-subscription-123"))
410+
{
411+
'primaryKey': '<redacted>',
412+
'secondaryKey': '<redacted>'
413+
}
414+
"""
415+
416+
url = (f"https://management.azure.com/subscriptions/{self._apim_subscription_id}/resourceGroups/"
417+
f"{self._apim_rg_name}/providers/Microsoft.ApiManagement/service/{self._apim_name}/"
418+
f"subscriptions/{user_id}/listSecrets"
419+
f"?api-version=2022-08-01")
420+
421+
headers = self._get_auth_headers()
422+
423+
response = requests.post(url, headers=headers)
424+
425+
if response.status_code == 200:
426+
return response.json()
427+
else:
428+
raise APIMSubscriptionNotFoundError(
429+
f"Subscription for user with id {user_id} not found. Status code: {response.status_code}, "
430+
f"Response: {response.text}")

0 commit comments

Comments
 (0)