My timetagger runfile

timetaggerRun.py

"""
Basic script to run timetagger.
You can use this to run timetagger locally. If you want to run it
online, you'd need to take care of authentication.
"""

import logging
from pkg_resources import resource_filename

import asgineer
from timetagger import config
from timetagger.server import (
    authenticate,
    AuthException,
    api_handler_triage,
    get_webtoken_unsafe,
    create_assets_from_dir,
    enable_service_worker,
)

config.bind = "0.0.0.0:8888"
logger = logging.getLogger("asgineer")

# Get sets of assets provided by TimeTagger
common_assets = create_assets_from_dir(resource_filename("timetagger.common", "."))
apponly_assets = create_assets_from_dir(resource_filename("timetagger.app", "."))
image_assets = create_assets_from_dir(resource_filename("timetagger.images", "."))
page_assets = create_assets_from_dir(resource_filename("timetagger.pages", "."))

# Combine into two groups. You could add/replace assets here.
app_assets = dict(**common_assets, **image_assets, **apponly_assets)
web_assets = dict(**common_assets, **image_assets, **page_assets)

# Enable the service worker so the app can be used offline and is installable
enable_service_worker(app_assets)

# Turn asset dicts into handlers. This feature of Asgineer provides
# lightning fast handlers that support compression and HTTP caching.
app_asset_handler = asgineer.utils.make_asset_handler(app_assets, max_age=0)
web_asset_handler = asgineer.utils.make_asset_handler(web_assets, max_age=0)


@asgineer.to_asgi
async def main_handler(request):
    """
    The main handler where we delegate to the API or asset handler.
    We serve at /timetagger for a few reasons, one being that the service
    worker won't interfere with other stuff you might serve on localhost.
    """
    
    if request.path == "/" or request.path == "/app" or request.path.startswith("/app/demo") or request.path.startswith("/app/sandbox"):
        return 307, {"Location": "/app/"}, b""  # Redirec

    if request.path.startswith("/api/v2/"):
        path = request.path[8:].strip("/")
        return await api_handler(request, path)
    elif request.path.startswith("/app/"):
        path = request.path[5:].strip("/")
        return await app_asset_handler(request, path)
    else:
        path = request.path[1:].strip("/")
        return await web_asset_handler(request, path)


async def api_handler(request, path):
    # Some endpoints do not require authentication
    if not path and request.method == "GET":
        return 200, {}, "See https://timetagger.readthedocs.io"
    elif path == "webtoken_for_localhost" or path == "bootstrap_authentication":
        return await webtoken_for_localhost(request)

    # Authenticate and get user db
    try:
        auth_info, db = await authenticate(request)
        if 'x-authentik-username' in request.headers and auth_info["username"] != request.headers['x-authentik-username']:
            raise AuthException("User changed")

    except AuthException as err:
        return 401, {}, f"Please login again: {err}"

    # Handle endpoints that require authentication
    return await api_handler_triage(request, path, auth_info, db)


async def webtoken_for_localhost(request):
    # Establish that we can trust the client
    if 'x-authentik-username' not in request.headers:
        return 403, {}, "forbidden: must be on authenticated"
    remoteUser = request.headers['x-authentik-username']
    if not remoteUser:
        return 403, {}, "forbidden: must be on authenticated"

    # Return the webtoken for the default user
    token = await get_webtoken_unsafe(remoteUser)
    return 200, {}, dict(token=token)


if __name__ == "__main__":
    asgineer.run(main_handler, "uvicorn", config.bind, log_level="warning")