Code cleanup

api.py

@@ -1,11 +1,17 @@
-import random, string, sys
+import random
+import string
+import sys
+
 import requests
 
+s = requests.session()
+
+
 def __post_request(url, json_data):
     api_url = f"{api_host}/{url}"
     headers = {'X-API-Key': api_key, 'Content-type': 'application/json'}
 
-    req = requests.post(api_url, headers=headers, json=json_data)
+    req = s.post(api_url, headers=headers, json=json_data)
     rsp = req.json()
     req.close()
 
@@ -14,24 +20,25 @@ def __post_request(url, json_data):
 
     if not "type" in rsp or not "msg" in rsp:
         sys.exit(f"API {url}: got response without type or msg from Mailcow API")
-    
+
     if rsp['type'] != 'success':
         sys.exit(f"API {url}: {rsp['type']} - {rsp['msg']}")
 
+
 def add_user(email, name, active, quotum):
     password = ''.join(random.choices(string.ascii_letters + string.digits, k=20))
     json_data = {
-        'local_part':email.split('@')[0],
-        'domain':email.split('@')[1],
-        'name':name,
-        'quota':str(quotum),
-        'password':password,
-        'password2':password,
+        'local_part': email.split('@')[0],
+        'domain': email.split('@')[1],
+        'name': name,
+        'quota': str(quotum),
+        'password': password,
+        'password2': password,
         "active": 1 if active else 0
     }
 
     __post_request('api/v1/add/mailbox', json_data)
-    
+
     json_data = {
         'items': [email],
         'attr': {
@@ -41,25 +48,26 @@ def add_user(email, name, active, quotum):
                 "spam_score",
                 "spam_policy",
                 "delimiter_action",
-                #"syncjobs",
-                #"eas_reset",
+                # "syncjobs",
+                # "eas_reset",
                 "quarantine",
-                #"sogo_profile_reset",
-                #"quarantine_attachments",
+                # "sogo_profile_reset",
+                # "quarantine_attachments",
                 "quarantine_notification",
-                #"app_passwds",
+                # "app_passwds",
                 "pushover"
             ]
         }
     }
-    
+
     __post_request('api/v1/edit/user-acl', json_data)
 
+
 def edit_user(email, active=None, name=None):
     attr = {}
-    if (active is not None):
+    if active is not None:
         attr['active'] = 1 if active else 0
-    if (name is not None):
+    if name is not None:
         attr['name'] = name
 
     json_data = {
@@ -69,25 +77,27 @@ def edit_user(email, active=None, name=None):
 
     __post_request('api/v1/edit/mailbox', json_data)
 
+
 def __delete_user(email):
     json_data = [email]
 
     __post_request('api/v1/delete/mailbox', json_data)
 
+
 def check_user(email):
     url = f"{api_host}/api/v1/get/mailbox/{email}"
     headers = {'X-API-Key': api_key, 'Content-type': 'application/json'}
-    req = requests.get(url, headers=headers)
+    req = s.get(url, headers=headers)
     rsp = req.json()
     req.close()
-    
+
     if not isinstance(rsp, dict):
         sys.exit("API get/mailbox: got response of a wrong type")
 
-    if (not rsp):
-        return (False, False, None)
+    if not rsp:
+        return False, False, None
 
     if 'active_int' not in rsp and rsp['type'] == 'error':
         sys.exit(f"API {url}: {rsp['type']} - {rsp['msg']}")
-    
-    return (True, bool(rsp['active_int']), rsp['name'])
+
+    return True, bool(rsp['active_int']), rsp['name']

filedb.py

@@ -1,52 +1,56 @@
-import datetime, os
-
+import datetime
 import logging
-logging.basicConfig(format='%(asctime)s %(message)s', datefmt='%d.%m.%y %H:%M:%S', level=logging.INFO)
-
-import sqlalchemy
+import os
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy import create_engine, Column, String, Boolean, DateTime
 from sqlalchemy.orm import sessionmaker
 
-db_file = 'db/ldap-mailcow.sqlite3'
+logging.basicConfig(format='%(asctime)s %(message)s', datefmt='%d.%m.%y %H:%M:%S', level=logging.INFO)
 
+db_file = 'db/ldap-mailcow.sqlite3'
 Base = declarative_base()
 
-class DbUser(Base): # type: ignore
+
+class DbUser(Base):  # type: ignore
     __tablename__ = 'users'
     email = Column(String, primary_key=True)
     active = Column(Boolean, nullable=False)
     last_seen = Column(DateTime, nullable=False)
 
+
 Session = sessionmaker()
 
 if not os.path.isfile(db_file):
-    logging.info (f"New database file created: {db_file}")
+    logging.info(f"New database file created: {db_file}")
 
-db_engine = create_engine(f"sqlite:///{db_file}") # echo=True
+db_engine = create_engine(f"sqlite:///{db_file}")  # echo=True
 Base.metadata.create_all(db_engine)
 Session.configure(bind=db_engine)
 session = Session()
 session_time = datetime.datetime.now()
 
+
 def get_unchecked_active_users():
     query = session.query(DbUser.email).filter(DbUser.last_seen != session_time).filter(DbUser.active == True)
 
     return [x.email for x in query]
 
+
 def add_user(email, active=True):
     session.add(DbUser(email=email, active=active, last_seen=session_time))
     session.commit()
 
+
 def check_user(email):
     user = session.query(DbUser).filter_by(email=email).first()
     if user is None:
-        return (False, False)
+        return False, False
     user.last_seen = session_time
     session.commit()
-    return (True, user.active)
+    return True, user.active
+
 
 def user_set_active_to(email, active):
     user = session.query(DbUser).filter_by(email=email).first()
     user.active = active
-    session.commit()
+    session.commit()