main.py

#!/usr/bin/env python3
#  Copyright (c) 2022 - 2026 Ricardo Bartels. All rights reserved.
#
#  wordpress-hash-event-api
#
#  This work is licensed under the terms of the MIT license.
#  For a copy, see file LICENSE.txt included in this
#  repository or visit: <https://opensource.org/licenses/MIT>.

import uvicorn
import os
import logging
import json
import psutil

from fastapi import FastAPI
from starlette.responses import RedirectResponse
from contextlib import asynccontextmanager

from config.models.api import APIConfigSettings
from config.models.app import AppSettings
from config.models.calendar import CalendarConfigSettings
from config.models.database import DBSettings
from listmonk.handler import ListMonkHandler, ListMonkSettings
from config.api import BasicAPISettings
import config
from api.security import api_key_valid, set_api_key
from api.routers import runs, send_newsletter
from source.database import setup_db_handler
from source.manage_event_fields import update_event_manager_fields
from common.log import get_logger

config_file_name = "config.ini"
logging_config_file_name = "log-config.json"
default_log_level = "INFO"


def get_app() -> FastAPI:

    basic_api_settings = BasicAPISettings()

    # get config file path
    config_file = config.get_config_file(config_file_name)

    log = get_logger()

    config_handler = None
    if os.path.exists(config_file):

        config_handler = config.open_config_file(config_file)

    log.info(f"Starting {basic_api_settings.description} v{basic_api_settings.version}")

    if not os.path.exists(config_file):
        log.warning(f"Config file '{config_file}' not found. Reading config from env vars")
    elif config_handler is None:
        log.warning(f"Problems while reading config file. Reading config from env vars")

    if log.getEffectiveLevel() <= logging.DEBUG:
        basic_api_settings.debug = True

    # parse settings for db and initialize db connection
    db_settings = config.get_config_object(config_handler, DBSettings)
    conn = setup_db_handler(
        host_name=db_settings.host,
        user_name=db_settings.username,
        user_password=db_settings.password,
        db_name=db_settings.name,
        db_port=db_settings.port
    )

    if conn is None or conn.session is None:
        log.error("Exit due to database connection error")
        exit(1)

    log.info("Database connection successfully started")

    # read app settings from config and try to find settings in WordPress db if not defined in config
    app_settings = config.get_config_object(config_handler, AppSettings)

    # try to find further settings in DB if undefined
    for key, value in app_settings:
        if value is None:
            db_setting = conn.get_config_item(key)
            if db_setting is not None:
                log.debug(f"Config: updating {AppSettings.config_section_name()}.{key} = {db_setting}")
                value = db_setting

        setattr(app_settings, key, value)

    # parse settings
    config.app_settings = config.validate_config_object(AppSettings, app_settings.model_dump())

    # update event manager fields in database
    update_event_manager_fields()

    # get calendar settings
    config.calendar_settings = config.get_config_object(config_handler, CalendarConfigSettings)

    # initialize listmonk
    listmonk_settings = config.get_config_object(config_handler, ListMonkSettings)

    if listmonk_settings.enabled is True:
        ListMonkHandler(listmonk_settings)

    # initialize FastAPI app
    # parse api settings from config
    api_settings = config.get_config_object(config_handler, APIConfigSettings)
    if api_settings.root_path is not None:
        basic_api_settings.root_path = api_settings.root_path

    # set api key if defined
    set_api_key(api_settings.token)

    # close DB connection on shutdown
    @asynccontextmanager
    async def lifespan(_: FastAPI):
        yield
        if conn is not None:
            conn.close()

    # create FastAPI instance
    server = FastAPI(**{**basic_api_settings.model_dump(), "lifespan": lifespan})

    # disable API authorization if no token is defined
    if api_settings.token is None:
        log.info("No API token defined, disabling API Authentication")
        server.dependency_overrides[api_key_valid] = lambda: None

    # add default route redirect to docs
    @server.get("/", include_in_schema=False)
    def redirect_to_docs() -> RedirectResponse:
        return RedirectResponse(f"{basic_api_settings.root_path}/docs")

    @server.get("/status", include_in_schema=False)
    def status():
        return {"status": "ok"}

    # add runs routes
    server.include_router(runs.router_runs)

    if listmonk_settings.enabled is True:
        # add newsletter post route
        server.include_router(send_newsletter.newsletter)

    return server


app = get_app()


def run():

    reload = False
    logging_config_location = os.path.sep.join([os.path.dirname(__file__), "config", logging_config_file_name])

    try:
        with open(logging_config_location, "r") as f:
            data = json.load(f)
    except Exception as e:
        print(f"ERROR: unable to open logging configuration: {e}")
        exit(1)

    # read log level from env
    log_level_from_env = os.getenv("LOG_LEVEL", default_log_level).lower()
    if log_level_from_env in ("debug", "trace"):
        reload = True

    if len(log_level_from_env) == 0:
        log_level_from_env = default_log_level.lower()

    # if process was initiated by systemd then remove the timestamp from log output
    try:
        if psutil.Process(psutil.Process(os.getpid()).ppid()).name() == "systemd":
            data["formatters"]["default"]["fmt"] = data["formatters"]["default"]["fmt"].replace('%(asctime)s - ', "")
            data["formatters"]["access"]["fmt"] = data["formatters"]["access"]["fmt"].replace('%(asctime)s - ', "")
    except Exception as e:
        print(f"unable to determine parent process name: {e}")

    uvicorn.run("main:app",
                host="0.0.0.0",
                port=8000,
                reload=reload,
                log_level=log_level_from_env,
                log_config=data,
                proxy_headers=True)


if __name__ == '__main__':
    run()

# EOF