API Architecture

This document describes the structure of the Flask-based API, including run.py, the api.<domain> packages, and the endpoint and helper modules such as api.genealogy.persons and api.genealogy.persons_support.

Contents

1. Overview
2. Directory structure
3. Top-level app: run.py
4. Domain packages: api/<domain>/__init__.py
5. Endpoint modules: api/<domain>/xxx.py
6. Helper / support modules
7. How to add a new domain
8. How to add a new endpoint

1. Overview

The project exposes a Flask-based API with a dynamic, modular structure:

• A single top-level Flask app (run.py) that:
  • Creates and configures the Flask application
  • Configures OAuth, sessions, and CORS
  • Auto-discovers api.* packages
  • Calls each package's register_api(app)
  • Scans all routes and writes a machine-readable registry (registered_apis.json)
  • Exposes helper endpoints like /registered_apis and /docs
• A set of domain packages under ./api/ (e.g. api.genealogy, api.k3, api.video_server), each responsible for a logical area of the system.
• Inside each domain package, one or more endpoint modules (e.g. api/genealogy/persons.py) that declare routes via a simple contract:

  def register_api(bp: Blueprint):
      ...
  

• Optional support / helper modules (e.g. api.genealogy.support, api.genealogy.persons_support) holding shared logic and DB helpers.

Thanks to auto-discovery, new domains and new endpoints can be added without modifying run.py.

2. Directory structure

Conceptual directory layout:

project/
├─ run.py                     # Main Flask app + auto-registration + docs
├─ registered_apis.json       # Auto-generated registry of endpoints
└─ api/
   ├─ genealogy/
   │  ├─ __init__.py          # Blueprint + auto-import of genealogy endpoints
   │  ├─ persons.py           # /genealogy/persons endpoint(s)
   │  ├─ persons_support.py   # helper functions used by persons.py
   │  ├─ support.py           # shared DB helpers (e.g. fetch_one, fetch_all)
   │  └─ ...                  # other genealogy-related endpoints
   ├─ k3/
   │  ├─ __init__.py
   │  └─ ...
   └─ video_server/
      ├─ __init__.py
      └─ ...

Every api/<domain> folder is a Python package with:

• __init__.py → defines a Blueprint and auto-registers its endpoint modules.
• *.py → endpoint modules that implement HTTP handlers.
• *_support.py (optional) → helper/service modules used by those endpoints.

3. Top-level app: run.py

3.1 Flask, OAuth, CORS and sessions

run.py is responsible for creating and configuring the Flask app:

• Creates the Flask application instance and sets secret_key.
• Initializes Authlib's OAuth() with the app.
• Configures session cookie behaviour (important for OAuth redirects).
• Enables CORS for specific origins on rforssen.net, with support for credentials.

3.2 Import paths

To make sure imports like import api.genealogy work consistently, run.py adds:

• The project root, and
• The parent directory of api

to sys.path.

3.3 Discovering api.* packages

Using pkgutil.walk_packages, run.py scans the ./api directory and collects all packages into API_PACKAGES (e.g. api.genealogy, api.k3, api.video_server).

3.4 registered_tree – internal registration tree

run.py maintains a nested dictionary called registered_tree that records:

• Which packages exist under api.
• For each package:
  • whether register_api(app) was found and called,
  • whether it was skipped (no register_api), or
  • whether an error occurred on import/registration.
• For each module under api, which endpoints it owns and which HTTP methods they support.

Later, this tree is stored inside registered_apis.json.

3.5 Calling register_api(app) on each package

For each discovered api.<domain> package:

1. run.py imports it via importlib.import_module(pkg_name).
2. If the package has a register_api(app) function, run.py calls it.
3. The package then:
   • creates a Blueprint,
   • auto-discovers its endpoint modules,
   • lets them attach routes via register_api(bp), and
   • registers the Blueprint on the app.

If a package has no register_api, or an exception occurs, this is recorded in registered_tree with a status like "skipped" or "error".

3.6 Route inspection

After all packages have registered their routes, run.py inspects Flask's URL map:

• Iterates over app.url_map.iter_rules().
• Skips the static endpoint.
• For each route, it records:
  • URL rule (e.g. /genealogy/persons)
  • endpoint name
  • HTTP methods
  • Python module name
  • file path

Routes are grouped by module and inserted into the appropriate node of registered_tree so that each api.<domain>.<module> lists its own endpoints.

3.7 registered_apis.json and helper endpoints

run.py writes a JSON file, registered_apis.json, containing:

• A timestamp.
• registered_apis: a hierarchical view of packages and endpoints.
• flask_routes: a flat list of all routes with metadata.

It also exposes a few helper endpoints:

• GET / — Simple health check returning "Flask API is running.".
• GET /registered_apis — Returns the contents of registered_apis.json.
• GET /docs — Builds a minimal OpenAPI-like document from the logged routes. This can be used as input for Swagger UI or other tooling.

4. Domain packages: api/<domain>/__init__.py

Each domain package (for example api.genealogy) defines:

1. A Blueprint with a URL prefix.
2. A function register_api(app) that:
   • auto-discovers endpoint modules within the package,
   • calls their register_api(bp) function when present,
   • registers the Blueprint on the Flask application.

Example: api/genealogy/__init__.py

from flask import Blueprint
import importlib
import pkgutil
import os

bp = Blueprint("genealogy", __name__, url_prefix="/genealogy")

def register_api(app):
    package_path = os.path.dirname(__file__)
    package_name = __name__  # 'api.genealogy'

    for _, module_name, is_pkg in pkgutil.iter_modules([package_path]):
        full_module_name = f"{package_name}.{module_name}"
        try:
            mod = importlib.import_module(full_module_name)
            if hasattr(mod, "register_api"):
                mod.register_api(bp)
        except Exception as e:
            print(f"⚠️ Failed to import {full_module_name}: {e}")

    app.register_blueprint(bp)

All modules under api.genealogy that implement register_api(bp) automatically attach their routes to this Blueprint and are exposed at URLs starting with /genealogy/....

5. Endpoint modules: api/<domain>/xxx.py

Each endpoint module is responsible for a set of related HTTP routes. The only requirement is that it exposes a function:

def register_api(bp: Blueprint):
    # define one or more routes on bp

Inside this function, routes are bound to the domain Blueprint with @bp.route(...).

5.1 Example: api.genealogy.persons

The persons.py module defines the /genealogy/persons endpoint, using helper functions moved to api.genealogy.persons_support.

from flask import Blueprint, request, jsonify
from api.genealogy.support import fetch_one, fetch_all, getGenDBconnection
from api.genealogy.persons_support import generate_columns_str, getFamilies, getReadAlso, getFamObj

print("✅ persons.py loaded")

def register_api(bp: Blueprint):

    @bp.route("/persons")
    def persons_handler():
        log = []
        action = request.args.get("action")
        id = request.args.get("id")

        if action != "onload":
            return jsonify({"error": f"Action {action} not implemented", "log": log}), 400

        conn = getGenDBconnection(log)
        if conn is None:
            return jsonify({"error": "Database connection failed", "log": log}), 500

        cursor = conn.cursor(dictionary=True, buffered=True)

        response = {
            "log": log,
            "person": {},
            "childhoodFamily": {},
            "families": [],
            "notes": [],
            "ancestorRelation": []
        }

        try:
            response["person"] = fetch_one(cursor,
                                           "SELECT * FROM tblIndividual WHERE IndKey = %s",
                                           id, log)
            response["person"]["Facts"] = fetch_all(cursor,
                                                    "SELECT * FROM tblFacts WHERE IndFamKey = %s",
                                                    (id,), log)
            response["person"]["Fbok"] = fetch_one(cursor,
                                                   "SELECT * FROM tblFbok WHERE IndKey = %s",
                                                   id, log)
            response["person"]["Dbok"] = fetch_one(cursor,
                                                   "SELECT * FROM tblDbok WHERE IndKey = %s",
                                                   id, log)

            fam_register = fetch_one(cursor,
                                     "SELECT * FROM tblRelation WHERE IndKey = %s",
                                     id, log)
            if fam_register:
                response["childhoodFamily"] = getFamObj(fam_register["FamKey"], cursor)
                response["halfFamilies"] = {"mother": [], "father": []}

                if response["childhoodFamily"].get("father") \
                        and response["childhoodFamily"]["father"].get("IndKey"):
                    father_id = response["childhoodFamily"]["father"]["IndKey"]
                    halfFamilies = fetch_all(cursor,
                                             "SELECT * FROM tblFamily WHERE FamSpouse1 = %s",
                                             (father_id,), log)
                    log.append(halfFamilies)
                    for halfFamily in halfFamilies:
                        if halfFamily["FamKey"] != fam_register["FamKey"]:
                            response["halfFamilies"]["father"].append(
                                getFamObj(halfFamily["FamKey"], cursor))

                if response["childhoodFamily"].get("mother") \
                        and response["childhoodFamily"]["mother"].get("IndKey"):
                    mother_id = response["childhoodFamily"]["mother"]["IndKey"]
                    halfFamilies = fetch_all(cursor,
                                             "SELECT * FROM tblFamily WHERE FamSpouse2 = %s",
                                             (mother_id,), log)
                    log.append(halfFamilies)
                    for halfFamily in halfFamilies:
                        if halfFamily["FamKey"] != fam_register["FamKey"]:
                            response["halfFamilies"]["mother"].append(
                                getFamObj(halfFamily["FamKey"], cursor))

            response["notes"] = fetch_all(cursor,
                                          "SELECT * FROM tblNotes WHERE IndFamKey = %s",
                                          (id,), log)
            response["families"] = getFamilies(cursor, response["person"], log)
            response.update(getReadAlso(response["person"], cursor, log))

        except Exception as e:
            log.append(f"❌ Exception: {str(e)}")
            return jsonify({"error": "Query error", "log": log}), 500
        finally:
            conn.close()

        return jsonify(response)

Endpoint modules are focused on:

• Parsing request parameters.
• Opening DB connections using shared helpers.
• Calling domain-specific helper functions.
• Composing and returning JSON responses.

6. Helper / support modules

Helpers and shared logic are kept in separate files so that endpoint modules remain readable and easy to test. These modules generally do not import Flask; they operate purely on data.

6.1 Generic support: api.genealogy.support

This module typically contains generic DB utilities such as:

• getGenDBconnection(log) — Open a connection to the genealogy DB.
• fetch_one(cursor, sql, params, log) — Execute a query and fetch a single row.
• fetch_all(cursor, sql, params, log) — Execute a query and fetch all rows.

6.2 Persons-specific helpers: api.genealogy.persons_support

This module contains the helper functions used specifically by persons.py, moved out of the endpoint module to keep it thin:

• generate_columns_str(columns)
• getFamilies(cursor, person, log)
• getReadAlso(person, cursor, log)
• getFamObj(fam_key, cursor)

Example (simplified):

from api.genealogy.support import fetch_one, fetch_all

def generate_columns_str(columns):
    parts = []
    for col in columns:
        if "expression" in col:
            expr = col["expression"]
            alias = col.get("alias")
            parts.append(f"{expr} AS `{alias}`" if alias else expr)
        else:
            field = f"`{col['field']}`"
            alias = col.get("alias")
            parts.append(f"{field} AS `{alias}`" if alias else field)
    return ", ".join(parts)

# ... getFamilies, getReadAlso, getFamObj ...

These helpers encapsulate the SQL logic and result shaping, while the endpoint focuses on HTTP and JSON.

7. How to add a new domain

To add a new domain, for example api.example:

1. Create api/example/__init__.py with a Blueprint and a register_api(app) function following the same pattern as api.genealogy.__init__.py.
2. Add endpoint modules under api/example that implement register_api(bp).
3. Restart the Flask app.

run.py will automatically discover api.example and call register_api(app); no changes to run.py are required.

8. How to add a new endpoint

To add a new endpoint to an existing domain, for example /genealogy/status:

1. Create api/genealogy/status.py:

   from flask import Blueprint, jsonify
   
   def register_api(bp: Blueprint):
   
       @bp.route("/status", methods=["GET"])
       def status():
           return jsonify({"domain": "genealogy", "status": "ok"})
   

2. Restart the Flask app.

The new route will:

• Be available at /genealogy/status.
• Appear in registered_apis.json under the proper module.
• Be included in the /docs OpenAPI-like schema.