init

.gitignore

+.idea
+config/
+p.py
\ No newline at end of file

README.md

+# NACSOS Core
+This repository contains the core data management platform of NACSOS.
+It accesses the database via the `nacsos-data` package and exposes the functionality via an API.
+It also serves the web frontend.
+
+## Installation
+- Requires Python 3.9+, tested with Python 3.10.2
+
+```bash
+virtualenv venv
+source venv/bin/activate
+pip install -r requirements.txt
+``` 
+
+At the moment, there is a breaking bug in tap, so you have to edit `tap/utils.py` in line 182 after installation:
+```python
+def get_class_column(obj: type) -> int:
+    """Determines the column number for class variables in a class."""
+    first_line = 1
+    for token_type, token, (start_line, start_column), (end_line, end_column), line in tokenize_source(obj):
+        if token.strip() == '@':
+            first_line += 1
+        if start_line <= first_line or token.strip() == '':
+            continue
+
+        return start_column
+```
+
+For development, it is advised to install `nacsos-data` locally (not from git) via
+```bash
+pip install -e ../nacsos-data/
+```
+(assuming both projects live side-by-side)
+
+## Running the database with docker
+Start up the database by running docker (or use your local instance)
+```bash
+sudo systemctl start docker
+docker-compose up
+```
+
+## Starting the server
+```bash
+python main.py
+
+# optionally, you can specify the config file directly
+NACSOS_CONFIG=config/local.toml python main.py
+
+# additionally, you can always override all exposed config variables directly
+# for more info, check
+python main.py -h
+```
+
+The configuration is read in the following order (and overridden by consecutive steps):
+- `@dataclasses` in `server/util/config.py`
+- TOML config file (either `config/default.toml` or whatever is in `NACSOS_CONFIG`)
+- Command line arguments
+
+The default config is set up to work with a locally running docker instance with its respective default config.
+It should never be changed, always make a local copy and never commit it to the repository!
\ No newline at end of file

config/default.toml

+[server]
+host = "localhost"
+port = 8080
+
+[db]
+host = "nacsos_postgres"
+port = 5432
+user = "root"
+pw = "root"
+database = "nacsos_core"
\ No newline at end of file

main.py

+from fastapi import FastAPI
+
+app = FastAPI()
+
+
+@app.get("/")
+async def root():
+    return {"message": "Hello World"}
+
+
+@app.get("/hello/{name}")
+async def say_hello(name: str):
+    return {"message": f"Hello {name}"}

requirements.txt

+fastapi==0.75.0
+typed-argument-parser==1.7.2
+uvicorn~=0.17.6
+toml==0.10.2
\ No newline at end of file

server/main.py

+#!/usr/bin/env python3
+
+from common import init_logging, init_config
+
+
+def run(args=None):
+    init_config(args)
+
+    from data.database import init_db
+    from api import Server
+
+    init_logging()
+    server = Server()
+    init_db(server.app)
+
+    return server
+
+
+if __name__ == '__main__':
+    run()
\ No newline at end of file

server/util/config.py

+from tap import Tap
+from tap.utils import get_class_variables
+from typing import get_type_hints, Any
+from dataclasses import dataclass
+from collections import OrderedDict
+import os
+import toml
+
+NestedConfigDict = dict[str, dict[str, Any]]
+
+
+@dataclass
+class ServerConfig:
+    host: str = 'localhost'  # host to run this server on
+    port: int = 8080  # port for this serve to listen at
+
+
+@dataclass
+class DatabaseConfig:
+    pw: str  # password for the database user
+    user: str = 'nacsos'  # username for the database
+    database: str = 'nacsos_core'  # name of the database
+    host: str = 'localhost'  # host of the db server
+    port: int = 5432  # port of the db server
+
+
+class Config:
+    _sub_configs = {
+        'server': ServerConfig,
+        'db': DatabaseConfig,
+    }
+
+    DEFAULT_CONFIG_FILE = 'config/default.toml'
+
+    def __init__(self):
+        stored_config = self._read_config_file()
+        shlex_config = dict2shlex(stored_config)
+        config = self._read_cli_args(shlex_config)
+
+        self.server: ServerConfig = ServerConfig(**config['server'])
+        self.db: DatabaseConfig = DatabaseConfig(**config['db'])
+
+    def _read_config_file(self) -> NestedConfigDict:
+        conf_file = os.environ.get('NACSOS_CONF', self.DEFAULT_CONFIG_FILE)
+        with open(conf_file, 'r') as f:
+            return toml.load(f)
+
+    def _read_cli_args(self, shlex_config: list[str]):
+        """
+        This method generates a Tap (typed-argument-parser) instance from the config classes
+        and exposes the variables including help (comments) and types to the command line.
+        It then parses all CLI arguments and returns a nested dictionary.
+        :return:
+        """
+
+        # create a typed argument parser by gathering all sub-configs (as argument prefixes)
+        # to expose all config attributes to the command line
+        class ProgrammaticArgumentParser(Tap):
+            def configure(self):
+                class_variables = []
+                annotations = []
+                for cls_name, cls in Config._sub_configs.items():
+                    # append all class attributes, including annotations (e.g. comments)
+                    class_variables += [(f'{cls_name}_{var}', data)
+                                        for var, data in get_class_variables(cls).items()]
+
+                    # append all annotations (e.g. type hints)
+                    annotations += [(f'{cls_name}_{var}', data)
+                                    for var, data in get_type_hints(cls).items()]
+
+                    # transfer default parameters to this instance
+                    for var in get_type_hints(cls).keys():
+                        try:
+                            setattr(self, f'{cls_name}_{var}', getattr(cls, var))
+                        except AttributeError:
+                            # this attribute has no default value
+                            pass
+
+                # inject the gathered class variables and annotations to the tap instance
+                self.class_variables = OrderedDict(class_variables)
+                self._annotations = dict(annotations)
+                self.args_from_configs = shlex_config
+
+        # parse command line arguments
+        args = ProgrammaticArgumentParser(underscores_to_dashes=True).parse_args()
+
+        config = {}
+        for arg, value in args.as_dict().items():
+            parts = arg.split('_')
+            if parts[0] not in config:
+                config[parts[0]] = {}
+            config[parts[0]]['_'.join(parts[1:])] = value
+
+        return config
+
+
+def dict2shlex(config: NestedConfigDict) -> list[str]:
+    ret = []
+    for cls_name, attrs in config.items():
+        for attr, value in attrs.items():
+            ret.append(f'--{cls_name}-{attr.replace("_", "-")}')
+            ret.append(f'"{value}"')
+    return ret
+
+__all__ = []
\ No newline at end of file

server/util/log.py

+import configparser
+import argparse
+import logging
+import logging.config
+import yaml
+import math
+import traceback
+from uvicorn.logging import AccessFormatter, DefaultFormatter
+
+
+def init_logging(logger_name: str = None):
+    logging.config.dictConfig(get_logger_config())
+    if logger_name:
+        return logging.getLogger(logger_name)
+
+
+class AccessLogFormatter(AccessFormatter):
+    def formatMessage(self, record):
+        try:
+            record.__dict__.update({
+                'wall_time': record.__dict__['scope']['timing_stats']['wall_time'],
+                'cpu_time': record.__dict__['scope']['timing_stats']['cpu_time']
+            })
+        except KeyError:
+            record.__dict__.update({
+                'wall_time': '0.0?s',
+                'cpu_time': '0.0?s'
+            })
+        return super().formatMessage(record)
+
+
+class ColourFormatter(DefaultFormatter):
+    def formatMessage(self, record):
+        pad = (8 - len(record.levelname)) / 2
+        levelname = ' ' * math.ceil(pad) + record.levelname + ' ' * math.floor(pad)
+        if self.use_colors:
+            record.__dict__['levelnamec'] = self.color_level_name(levelname, record.levelno)
+        else:
+            record.__dict__['levelnamec'] = levelname
+
+        return super().formatMessage(record)
+
+
+def except2str(e, logger=None):
+    if config.getboolean('server', 'debug_mode'):
+        tb = traceback.format_exc()
+        if logger:
+            logger.error(tb)
+        return tb
+    return f'{type(e).__name__}: {e}'