forge/authentik-listmonk-bridge
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages Wiki Activity Actions

main.py: better exception handling and logging

Browse source
This commit is contained in:
Antoine Martin 2024-06-07 16:50:28 -04:00
parent e691aff300
commit 8fc8a79cf4
Signed by: forge
GPG key ID: D62A472A4AA7D541
1 changed files with 74 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

100
main.py
View file

@ -102,23 +102,33 @@ def listmonk_get_list(list):
for group_info in authentik_get_groups(): for group_info in authentik_get_groups():
try: try:
list_info = listmonk_get_list('group-' + group_info['name']) list_info = listmonk_get_list('group-' + group_info['name'])
except:
except Exception as e:
print('> Failed to find group as list, creating group-' + group_info['name']) print('> Failed to find group as list, creating group-' + group_info['name'])
create_resp = listmonk_create_list('group-' + group_info['name']) try:
print(create_resp) create_resp = listmonk_create_list('group-' + group_info['name'])
continue except:
print('> Failed to create group-' + group_info['name'])
print(create_resp)
finally:
if debug:
print(create_resp)
continue
# sync that apps exists as lists # sync that apps exists as lists
for app_info in authentik_get_apps(): for app_info in authentik_get_apps():
try: try:
list_info = listmonk_get_list('app-' + app_info['slug']) list_info = listmonk_get_list('app-' + app_info['slug'])
except:
except Exception as e:
print('> Failed to find app as list, creating app-' + app_info['slug']) print('> Failed to find app as list, creating app-' + app_info['slug'])
create_resp = listmonk_create_list('app-' + app_info['slug']) try:
print(create_resp) create_resp = listmonk_create_list('app-' + app_info['slug'])
continue except:
print('> Failed to create app-' + app_info['slug'])
print(create_resp)
finally:
if debug:
print(create_resp)
continue
# ensure that users are synced # ensure that users are synced
for user_info in authentik_get_users(): for user_info in authentik_get_users():
@ -131,29 +141,51 @@ for user_info in authentik_get_users():
try: try:
subscriber_info = listmonk_get_subscriber(user_info['username']) subscriber_info = listmonk_get_subscriber(user_info['username'])
if debug:
print(subscriber_info)
print(user_info)
# if get user info fails, assume that we have to create a new one # if get user info fails, assume that we have to create a new one
# TOOD: only create if the error is indeed user-existence related # TOOD: only create if the error is indeed user-existence related
except Exception as e: except:
print('> Failed to find username, creating subscriber') print('> Failed to find username, creating subscriber')
create_resp = listmonk_create_subscriber(user_info['email'], user_info['username'], status=user_active) try:
subscriber_info = listmonk_get_subscriber(user_info['username']) create_resp = listmonk_create_subscriber(user_info['email'], user_info['username'], status=user_active)
print(create_resp) except:
print('> Subscriber ' + user_info['username'] + ' failed to be created')
print(create_resp)
finally:
print('> Subscriber ' + user_info['username'] + ' successfully created')
subscriber_info = listmonk_get_subscriber(user_info['username'])
if debug:
print(create_resp)
if debug:
print(subscriber_info)
print(user_info)
# checks if user email matches subscriber email # checks if user email matches subscriber email
if user_info['email'] != subscriber_info['email']: if user_info['email'] != subscriber_info['email']:
print('> Email check failed, updating') print('> Email check failed, updating')
set_resp = listmonk_set_subscriber(subscriber_info['id'], user_info['email'], user_info['username'], subscriber_info['status']) try:
print(set_resp) set_resp = listmonk_set_subscriber(subscriber_info['id'], user_info['email'], user_info['username'], subscriber_info['status'])
except:
print('> Failed to set email address of ' + user_info['username'])
print(set_resp)
finally:
print('> Email of ' + user_info['username'] + ' succesfully set as ' + user_info['email'])
if debug:
print(set_resp)
# check if user and subscriber info match as it relates to status # check if user and subscriber info match as it relates to status
if user_active != subscriber_info['status'] and subscriber_info['status'] != "blocklisted": if user_active != subscriber_info['status'] and subscriber_info['status'] != "blocklisted":
print('> User status check failed, updating') print('> User status check failed, updating')
set_resp = listmonk_set_subscriber(subscriber_info['id'], user_info['email'], user_info['username'], status=user_active) try:
print(set_resp) set_resp = listmonk_set_subscriber(subscriber_info['id'], user_info['email'], user_info['username'], status=user_active)
except:
print('> Failed to set status of ' + user_info['username'] + ' as ' + user_active)
print(set_resp)
finally:
print('> Status of ' + user_info['username'] + ' succesfully set as ' + user_active)
if debug:
print(set_resp)
# creates list of lists_id # creates list of lists_id
lists_id = [] lists_id = []
@ -176,13 +208,29 @@ for user_info in authentik_get_users():
for id in groups_id: for id in groups_id:
if id not in lists_id: if id not in lists_id:
print('> User lists check failed, updating') print('> User lists check failed, updating')
set_resp = listmonk_set_list([subscriber_info['id']], "add", groups_id, status="confirmed") try:
print(set_resp) set_resp = listmonk_set_list([subscriber_info['id']], "add", groups_id, status="confirmed")
break except:
print('> Adding to lists of ' + user_info['username'] + ' failed')
print(set_resp)
finally:
print('> List additions of ' + user_info['username'] + ' successfully set')
if debug:
print(set_resp)
break
# checks if lists_id is in list of group id # checks if lists_id is in list of group id
# this will fail is user is in a subcriber list, but not in respective group # this will fail is user is in a subcriber list, but not in respective group
for id in lists_id: for id in lists_id:
if id not in groups_id: if id not in groups_id:
print('> User groups check failed, updating') print('> User groups check failed, updating')
set_resp = listmonk_set_list([subscriber_info['id']], "remove", [id], status="") try:
set_resp = listmonk_set_list([subscriber_info['id']], "remove", [id], status="")
except:
print('> Removing from lists of ' + user_info['username'] + ' failed')
print(set_resp)
finally:
print('> List removals of ' + user_info['username'] + ' successfully set')
if debug:
print(set_resp)