Auth

mapswipe_workers/mapswipe_workers/auth.py

@@ -1,150 +1,74 @@
-#!/usr/bin/python3
-#
-# Author: B. Herfort, M. Reinmuth, 2017
-############################################
-
-import json
-
-import psycopg2
 import firebase_admin
-from firebase_admin import credentials
-from firebase_admin import db
-
-from mapswipe_workers.definitions import CONFIG_PATH
-from mapswipe_workers.definitions import SERVICE_ACCOUNT_KEY_PATH
+import psycopg2
+from firebase_admin import credentials, db
 
+from mapswipe_workers.definitions import CONFIG, SERVICE_ACCOUNT_KEY_PATH
 
-def load_config():
-    """
-    Loads the user configuration values.
 
-    Returns
-    -------
-    dictonary
-    """
-    with open(CONFIG_PATH) as f:
-        CONFIG = json.load(f)
-    return CONFIG
+def get_api_key(tileserver: str) -> str:
+    if tileserver == "custom":
+        return None
+    else:
+        return CONFIG["imagery"][tileserver]["api_key"]
 
 
-def get_api_key(tileserver):
-    CONFIG = load_config()
-    try:
-        if tileserver == 'custom':
-            return None
-        else:
-            return CONFIG['imagery'][tileserver]['api_key']
-    except KeyError:
-        print(
-                f'Could not find the API key for imagery tileserver '
-                f'{tileserver} in {CONFIG_PATH}.'
-                )
-        raise
-
-
-def get_tileserver_url(tileserver):
-    CONFIG = load_config()
-    try:
-        if tileserver == 'custom':
-            return None
-        else:
-            return CONFIG['imagery'][tileserver]['url']
-    except KeyError:
-        print('Could not find the url for imagery tileserver {} in {}.'.format(
-            tileserver,
-            CONFIG_PATH
-            ))
-        raise
-
-
-def init_firebase():
-    try:
-        # Is an App instance already initialized?
-        firebase_admin.get_app()
-    except ValueError:
-        cred = credentials.Certificate(SERVICE_ACCOUNT_KEY_PATH)
-        # Initialize the app with a service account, granting admin privileges
-        firebase_admin.initialize_app(cred)
+def get_tileserver_url(tileserver: str) -> str:
+    if tileserver == "custom":
+        return None
+    else:
+        return CONFIG["imagery"][tileserver]["url"]
 
 
-def firebaseDB():
+def firebaseDB() -> object:
     try:
         # Is an App instance already initialized?
         firebase_admin.get_app()
         # Return the imported Firebase Realtime Database module
         return db
     except ValueError:
         cred = credentials.Certificate(SERVICE_ACCOUNT_KEY_PATH)
-        config = load_config()
-        databaseName = config['firebase']['database_name']
-        databaseURL = f'https://{databaseName}.firebaseio.com'
+        databaseName = CONFIG["firebase"]["database_name"]
+        databaseURL = f"https://{databaseName}.firebaseio.com"
 
         # Initialize the app with a service account, granting admin privileges
-        firebase_admin.initialize_app(cred, {
-            'databaseURL': databaseURL
-        })
+        firebase_admin.initialize_app(cred, {"databaseURL": databaseURL})
 
         # Return the imported Firebase Realtime Database module
         return db
 
 
 class postgresDB(object):
+    """Helper calss for Postgres interactions"""
+
     _db_connection = None
     _db_cur = None
 
     def __init__(self):
-        CONFIG = load_config()
-        try:
-            host = CONFIG['postgres']['host']
-            port = CONFIG['postgres']['port']
-            dbname = CONFIG['postgres']['database']
-            user = CONFIG['postgres']['username']
-            password = CONFIG['postgres']['password']
-        except KeyError:
-            raise Exception(
-                    f'Could not load postgres credentials '
-                    f'from the configuration file'
-                    )
+        host = CONFIG["postgres"]["host"]
+        port = CONFIG["postgres"]["port"]
+        dbname = CONFIG["postgres"]["database"]
+        user = CONFIG["postgres"]["username"]
+        password = CONFIG["postgres"]["password"]
 
         self._db_connection = psycopg2.connect(
-                database=dbname,
-                user=user,
-                password=password,
-                host=host,
-                port=port
-                )
+            database=dbname, user=user, password=password, host=host, port=port,
+        )
 
     def query(self, query, data=None):
         self._db_cur = self._db_connection.cursor()
         self._db_cur.execute(query, data)
         self._db_connection.commit()
         self._db_cur.close()
 
-    def copy_from(
-            self,
-            f,
-            table,
-            columns
-            ):
+    def copy_from(self, f, table, columns):
         self._db_cur = self._db_connection.cursor()
-        self._db_cur.copy_from(
-                f,
-                table,
-                columns=columns
-                )
+        self._db_cur.copy_from(f, table, columns=columns)
         self._db_connection.commit()
         self._db_cur.close()
 
-    def copy_expert(
-            self,
-            sql,
-            file,
-            ):
+    def copy_expert(self, sql, file):
         self._db_cur = self._db_connection.cursor()
-        self._db_cur.copy_expert(
-                sql,
-                file,
-                )
+        self._db_cur.copy_expert(sql, file)
         self._db_connection.commit()
         self._db_cur.close()
 

mapswipe_workers/mapswipe_workers/definitions.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import logging.config
 import os
@@ -10,12 +11,25 @@
 )
 from mapswipe_workers.project_types.footprint.footprint_project import FootprintProject
 
+
+class CustomError(Exception):
+    pass
+
+
+def load_config(CONFIG_PATH) -> dict:
+    """Read the configuration file."""
+    with open(CONFIG_PATH) as f:
+        return json.load(f)
+
+
 ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
 
 CONFIG_DIR = os.path.abspath("/usr/share/config/mapswipe_workers/")
 
 CONFIG_PATH = os.path.join(CONFIG_DIR, "configuration.json")
 
+CONFIG = load_config(CONFIG_PATH)
+
 SERVICE_ACCOUNT_KEY_PATH = os.path.join(CONFIG_DIR, "serviceAccountKey.json")
 
 LOGGING_CONFIG_PATH = os.path.join(CONFIG_DIR, "logging.cfg")
@@ -36,7 +50,3 @@
 
 logging.config.fileConfig(fname=LOGGING_CONFIG_PATH, disable_existing_loggers=True)
 logger = logging.getLogger("Mapswipe Workers")
-
-
-class CustomError(Exception):
-    pass