Initial commit

modules/app.py

@@ -0,0 +1,22 @@
+from fastapi import FastAPI
+from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
+
+app = FastAPI(title="GarbageReminder API", docs_url=None, redoc_url=None, version="0.1")
+
+
+@app.get("/docs", include_in_schema=False)
+async def custom_swagger_ui_html():
+    return get_swagger_ui_html(
+        openapi_url=app.openapi_url,
+        title=f"{app.title} - Documentation",
+        swagger_favicon_url="/favicon.ico",
+    )
+
+
+@app.get("/redoc", include_in_schema=False)
+async def custom_redoc_html():
+    return get_redoc_html(
+        openapi_url=app.openapi_url,
+        title=f"{app.title} - Documentation",
+        redoc_favicon_url="/favicon.ico",
+    )

modules/database.py

@@ -0,0 +1,26 @@
+from typing import Any, Mapping
+
+from async_pymongo import AsyncClient, AsyncCollection, AsyncDatabase
+
+from modules.utils import config_get
+
+db_config: Mapping[str, Any] = config_get("database")
+
+if db_config["user"] is not None and db_config["password"] is not None:
+    con_string = "mongodb://{0}:{1}@{2}:{3}/{4}".format(
+        db_config["user"],
+        db_config["password"],
+        db_config["host"],
+        db_config["port"],
+        db_config["name"],
+    )
+else:
+    con_string = "mongodb://{0}:{1}/{2}".format(
+        db_config["host"], db_config["port"], db_config["name"]
+    )
+
+db_client = AsyncClient(con_string)
+db: AsyncDatabase = db_client.get_database(name=db_config["name"])
+
+col_entries: AsyncCollection = db.get_collection("entries")
+col_locations: AsyncCollection = db.get_collection("locations")

modules/extensions_loader.py

@@ -0,0 +1,53 @@
+from importlib.util import module_from_spec, spec_from_file_location
+from os import getcwd, path, walk
+from pathlib import Path
+
+# =================================================================================
+
+
+# Import functions
+# Took from https://stackoverflow.com/a/57892961
+def get_py_files(src):
+    cwd = getcwd()  # Current Working directory
+    py_files = []
+    for root, dirs, files in walk(src):
+        py_files.extend(
+            Path(f"{cwd}/{root}/{file}") for file in files if file.endswith(".py")
+        )
+    return py_files
+
+
+def dynamic_import(module_name, py_path):
+    try:
+        module_spec = spec_from_file_location(module_name, py_path)
+        module = module_from_spec(module_spec)  # type: ignore
+        module_spec.loader.exec_module(module)  # type: ignore
+        return module
+    except SyntaxError:
+        print(
+            f"Could not load extension {module_name} due to invalid syntax. Check logs/errors.log for details.",
+            flush=True,
+        )
+        return
+    except Exception as exp:
+        print(f"Could not load extension {module_name} due to {exp}", flush=True)
+        return
+
+
+def dynamic_import_from_src(src, star_import=False):
+    my_py_files = get_py_files(src)
+    for py_file in my_py_files:
+        module_name = Path(py_file).stem
+        print(f"Importing {module_name} extension...", flush=True)
+        imported_module = dynamic_import(module_name, py_file)
+        if imported_module != None:
+            if star_import:
+                for obj in dir(imported_module):
+                    globals()[obj] = imported_module.__dict__[obj]
+            else:
+                globals()[module_name] = imported_module
+        print(f"Successfully loaded {module_name} extension", flush=True)
+    return
+
+
+# =================================================================================

modules/scheduler.py

@@ -0,0 +1,3 @@
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+scheduler = AsyncIOScheduler()

modules/utils.py

@@ -0,0 +1,71 @@
+import logging
+from pathlib import Path
+from traceback import format_exc
+from typing import Any, Union
+
+from ujson import JSONDecodeError, dumps, loads
+
+logger = logging.getLogger(__name__)
+
+
+def json_load(filepath: Union[str, Path]) -> Any:
+    """Load json file
+
+    ### Args:
+        * filepath (`Union[str, Path]`): Path to input file
+
+    ### Returns:
+        * `Any`: Some json deserializable
+    """
+    with open(filepath, "r", encoding="utf8") as file:
+        try:
+            output = loads(file.read())
+        except JSONDecodeError:
+            logger.error(
+                "Could not load json file %s: file seems to be incorrect!\n%s",
+                filepath,
+                format_exc(),
+            )
+            raise
+        except FileNotFoundError:
+            logger.error(
+                "Could not load json file %s: file does not seem to exist!\n%s",
+                filepath,
+                format_exc(),
+            )
+            raise
+        file.close()
+    return output
+
+
+def json_write(contents: Union[list, dict], filepath: Union[str, Path]) -> None:
+    """Save contents into json file
+
+    ### Args:
+        * contents (`Union[list, dict]`): Some json serializable
+        * filepath (`Union[str, Path]`): Path to output file
+    """
+    try:
+        with open(filepath, "w", encoding="utf8") as file:
+            file.write(dumps(contents, ensure_ascii=False, indent=4))
+            file.close()
+    except Exception as exp:
+        logger.error("Could not save json file %s: %s\n%s", filepath, exp, format_exc())
+    return
+
+
+def config_get(key: str, *args: str) -> Any:
+    """Get value of the config key
+
+    ### Args:
+        * key (`str`): The last key of the keys path.
+        * *args (`str`): Path to key like: dict[args][key].
+
+    ### Returns:
+        * `Any`: Value of provided key
+    """
+    this_dict = json_load(Path("config.json"))
+    this_key = this_dict
+    for dict_key in args:
+        this_key = this_key[dict_key]
+    return this_key[key]