Customizer API

The Customizer API is a FastAPI server that runs the Customizer and exposes endpoints to upload and manage payloads, check the status of payload processing and run maintenance tasks.

Start the server with:

pip install "pyxecm[api,customizer]"
pyxecm-api

The interactive OpenAPI documentation is available at http://localhost:8000/api once the server is running. See API Settings for the available configuration options.

Application

app

API Implementation for the Customizer to start and control the payload processing.

Functions:

Name	Description
lifespan	Lifespan function for FASTAPI to handle the startup and shutdown process.
run_api	Start the FastAPI Webserver.

lifespan async

lifespan(app: FastAPI) -> AsyncGenerator

Lifespan function for FASTAPI to handle the startup and shutdown process.

Parameters:

Name	Type	Description	Default
app	FastAPI	The FastAPI application.	required

Source code in packages/pyxecm/src/pyxecm_api/app.py

@asynccontextmanager
async def lifespan(
    app: FastAPI,  # noqa: ARG001
) -> AsyncGenerator:
    """Lifespan function for FASTAPI to handle the startup and shutdown process.

    Args:
        app (FastAPI):
            The FastAPI application.

    """

    if os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT"):
        # Set up OpenTelemetry tracing for the worker
        resource = Resource.create({"service.name": "pyxecm-customizer-api"})
        provider = TracerProvider(resource=resource)
        processor = BatchSpanProcessor(OTLPSpanExporter())
        provider.add_span_processor(processor)

        trace.set_tracer_provider(provider)

    logger.debug("API settings -> %s", api_settings)

    with tracer.start_as_current_span("import_payloads"):
        if api_settings.import_payload:
            logger.info("Importing filesystem payloads...")

            # Base Payload
            import_payload(payload=api_settings.payload)

            # External Payload
            import_payload(payload_dir=api_settings.payload_dir, dependencies=True)

            # Optional Payload
            import_payload(payload_dir=api_settings.payload_dir_optional)

    logger.info("Starting maintenance page thread...")
    if api_settings.maintenance_page:
        run_maintenance_page()

    yield

    logger.info("Shutdown")
    PAYLOAD_LIST.stop_payload_processing()
    LOG_SPOOL.close()

run_api

run_api() -> None

Start the FastAPI Webserver.

Source code in packages/pyxecm/src/pyxecm_api/app.py

def run_api() -> None:
    """Start the FastAPI Webserver."""

    # Check if Temp and Log dir exists
    if not os.path.exists(api_settings.temp_dir):
        os.makedirs(api_settings.temp_dir)
    if not os.path.exists(api_settings.logfolder):
        os.makedirs(api_settings.logfolder)

    # Check if Logfile and exists and is unique
    if os.path.isfile(os.path.join(api_settings.logfolder, api_settings.logfile)):
        customizer_start_time = datetime.now(UTC).strftime(
            "%Y-%m-%d_%H-%M",
        )
        api_settings.logfile = "customizer_{}.log".format(customizer_start_time)

    # Configure Logging for uvicorn
    log_config = uvicorn.config.LOGGING_CONFIG

    # Stdout
    log_config["formatters"]["standard"] = {
        "()": "uvicorn.logging.DefaultFormatter",
        # The log format with uvicorn's colored level name.
        "fmt": LOG_FORMAT.replace("%(levelname)s", "%(levelprefix)s"),
        "datefmt": LOG_DATE_FORMAT,
        "use_colors": True,
    }

    log_config["formatters"]["logfile"] = {
        "()": "uvicorn.logging.DefaultFormatter",
        "fmt": LOG_FORMAT,
        "datefmt": LOG_DATE_FORMAT,
        "use_colors": True,
    }

    log_config["formatters"]["accesslog"] = {
        "()": "uvicorn.logging.AccessFormatter",
        "fmt": '%(client_addr)s - [%(asctime)s] "%(request_line)s" %(status_code)s',
        "datefmt": "%d/%b/%Y:%H:%M:%S %z",
        "use_colors": False,
    }

    log_config["formatters"]["default"] = {
        "()": "uvicorn.logging.DefaultFormatter",
        "fmt": "%(asctime)s %(levelprefix)s %(message)s",
        "datefmt": LOG_DATE_FORMAT,
        "use_colors": False,
    }

    log_config["handlers"]["console"] = {
        "formatter": "standard",
        "class": "logging.StreamHandler",
        "stream": "ext://sys.stdout",
    }

    log_config["handlers"]["file"] = {
        "formatter": "logfile",
        "class": "logging.FileHandler",
        "filename": os.path.join(api_settings.logfolder, api_settings.logfile),
        "mode": "a",
    }

    log_config["handlers"]["accesslog"] = {
        "formatter": "accesslog",
        "class": "logging.FileHandler",
        "filename": os.path.join(api_settings.logfolder, "access.log"),
        "mode": "a",
    }

    log_config["handlers"]["access"] = {
        "formatter": "accesslog",
        "class": "logging.StreamHandler",
        "stream": "ext://sys.stdout",
    }

    log_config["loggers"]["uvicorn.access"]["handlers"] = ["access", "accesslog"]

    log_config["loggers"]["root"] = {
        "level": api_settings.loglevel,
        "handlers": ["console", "file"],
    }

    logger.info("Starting processing thread...")
    PAYLOAD_LIST.run_payload_processing(concurrent=api_settings.concurrent_payloads)

    uvicorn.run(
        "pyxecm_api:app",
        host=api_settings.bind_address,
        port=api_settings.bind_port,
        workers=api_settings.workers,
        reload=api_settings.reload,
        proxy_headers=True,
        forwarded_allow_ips="*",
        log_config=log_config,
    )

Authentication

router

Utility library to handle the authentication with OTDS.

Functions:

Name	Description
login	Login using OTDS and return a token.
read_users_me	Get the current user.

login

login(
    form_data: Annotated[
        OAuth2PasswordRequestForm, Depends()
    ],
) -> JSONResponse

Login using OTDS and return a token.

Source code in packages/pyxecm/src/pyxecm_api/auth/router.py

@router.post("/token", tags=["auth"])
def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]) -> JSONResponse:
    """Login using OTDS and return a token."""

    url = api_settings.otds_url + "/otdsws/rest/authentication/credentials"

    payload = json.dumps(
        {"userName": form_data.username, "password": form_data.password},
    )
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json",
    }

    try:
        response = requests.request(
            "POST",
            url,
            headers=headers,
            data=payload,
            timeout=10,
        )
    except requests.exceptions.ConnectionError as exc:
        raise HTTPException(
            status_code=500,
            detail=f"{exc.request.url} cannot be reached",
        ) from exc

    if response.ok:
        response_data = json.loads(response.text)
    else:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    return JSONResponse(
        {
            "access_token": response_data["ticket"],
            "token_type": "bearer",
            "userId": response_data["userId"],
        },
    )

read_users_me async

read_users_me(
    current_user: Annotated[
        User, Depends(get_current_user)
    ],
) -> JSONResponse

Get the current user.

current_user (User): The current user.

Source code in packages/pyxecm/src/pyxecm_api/auth/router.py

@router.get("/users/me", tags=["auth"])
async def read_users_me(current_user: Annotated[User, Depends(get_current_user)]) -> JSONResponse:
    """Get the current user.

    current_user (User):
        The current user.

    """

    if "otadmins@otds.admin" in current_user.groups:
        return JSONResponse(current_user.model_dump())
    else:
        raise HTTPException(
            status_code=403,
            detail=f"User {current_user.id} is not authorized",
        )

models

Models for FastAPI.

Classes:

Name	Description
User	Model for users authenticated by OTDS.

User

Bases: BaseModel

Model for users authenticated by OTDS.

functions

Utility library to handle the authentication with OTDS.

Functions:

Name	Description
get_groups	Get the groups of the user.
get_current_user	Get the current user from OTDS and verify it.
get_authorized_user	Check if the user is authorized (member of the Group otadmin@otds.admin or should be a tenantAdmin user).
get_otcsticket	Check if the user is authorized (member of the Group otadmin@otds.admin).

get_groups

get_groups(response: dict, token: str) -> list

Get the groups of the user.

Parameters:

Name	Type	Description	Default
response	dict	description	required
token	str	description	required

Returns:

Name	Type	Description
list	list	description

Source code in packages/pyxecm/src/pyxecm_api/auth/functions.py

def get_groups(response: dict, token: str) -> list:
    """Get the groups of the user.

    Args:
        response (dict): _description_
        token (str): _description_

    Returns:
        list: _description_

    """

    headers = {
        "Accept": "application/json",
        "otdsticket": token,
    }
    url = api_settings.otds_url + "/otdsws/rest/users/" + response["user"]["id"] + "/memberof"

    response = requests.request("GET", url, headers=headers, timeout=5)
    if response.ok:
        response = json.loads(response.text)
        return [group["id"] for group in response.get("groups", [])]

    # Return empty list if request wasn't successful
    return []

get_current_user

get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    api_key: Annotated[str, Depends(apikey_header)],
) -> User

Get the current user from OTDS and verify it.

Source code in packages/pyxecm/src/pyxecm_api/auth/functions.py

def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)], api_key: Annotated[str, Depends(apikey_header)]
) -> User:
    """Get the current user from OTDS and verify it."""

    if api_settings.api_key is not None and api_key == api_settings.api_key:
        return User(
            id="api",
            full_name="API Key",
            groups=["otadmins@otds.admin"],
            is_admin=True,
            is_sysadmin=True,
            is_tenantadmin=False,
        )

    if token and token.startswith("*OTDSSSO*"):
        url = api_settings.otds_url + "/otdsws/rest/currentuser"
        headers = {
            "Accept": "application/json",
            "otdsticket": token,
        }
        response = requests.request("GET", url, headers=headers, timeout=2)

        if response.ok:
            user_data = json.loads(response.text)

            # Check if user is tenant admin
            tenant_admin = False
            for attr in user_data["user"].get("values", []):
                if attr.get("name") == "oTType" and "TenantAdminUser" in attr.get("values", []):
                    tenant_admin = True
                    break

            return User(
                id=user_data["user"]["id"],
                full_name=user_data["user"]["name"],
                groups=get_groups(user_data, token),
                is_admin=user_data["isAdmin"],
                is_sysadmin=user_data["isSysAdmin"],
                is_tenantadmin=tenant_admin,
            )

    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid authentication credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )

get_authorized_user async

get_authorized_user(
    current_user: Annotated[
        User, Depends(get_current_user)
    ],
) -> User

Check if the user is authorized (member of the Group otadmin@otds.admin or should be a tenantAdmin user).

Source code in packages/pyxecm/src/pyxecm_api/auth/functions.py

async def get_authorized_user(current_user: Annotated[User, Depends(get_current_user)]) -> User:
    """Check if the user is authorized (member of the Group otadmin@otds.admin or should be a tenantAdmin user)."""

    if "otadmins@otds.admin" not in current_user.groups and not current_user.is_tenantadmin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=f"User {current_user.id} is not authorized",
        )
    return current_user

get_otcsticket async

get_otcsticket(
    otcsticket: Annotated[str, Depends(otcsticket)],
) -> User

Check if the user is authorized (member of the Group otadmin@otds.admin).

Source code in packages/pyxecm/src/pyxecm_api/auth/functions.py

async def get_otcsticket(otcsticket: Annotated[str, Depends(otcsticket)]) -> User:
    """Check if the user is authorized (member of the Group otadmin@otds.admin)."""

    return otcsticket

Common

router

API Implementation for the Customizer to start and control the payload processing.

Functions:

Name	Description
redirect_to_api	Redirect from / to /api.
get_status	Get the status of the Customizer.
shutdown	Endpoint to end the application.
list_browser_automation_files	List all browser automation files.
get_browser_automation_file	Download the logfile for a specific payload.

redirect_to_api async

redirect_to_api(request: Request) -> RedirectResponse

Redirect from / to /api.

Returns:

Type	Description
RedirectResponse	None

Source code in packages/pyxecm/src/pyxecm_api/common/router.py

@router.get("/", include_in_schema=False)
async def redirect_to_api(request: Request) -> RedirectResponse:
    """Redirect from / to /api.

    Returns:
        None

    """
    return RedirectResponse(url=f"{request.url.path}api")

get_status async

get_status() -> CustomizerStatus

Get the status of the Customizer.

Source code in packages/pyxecm/src/pyxecm_api/common/router.py

@router.get(path="/status", name="Get Status")
async def get_status() -> CustomizerStatus:
    """Get the status of the Customizer."""

    df = PAYLOAD_LIST.get_payload_items()

    if df is None:
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
            detail="Payload list is empty.",
        )

    all_status = df["status"].value_counts().to_dict()

    return CustomizerStatus(
        version=2,
        customizer_duration=(all_status.get("running", None)),
        customizer_end_time=None,
        customizer_start_time=None,
        status_details=all_status,
        status="Running" if "running" in all_status else "Stopped",
        debug=df["log_debug"].sum(),
        info=df["log_info"].sum(),
        warning=df["log_warning"].sum(),
        error=df["log_error"].sum(),
        critical=df["log_critical"].sum(),
    )

shutdown

shutdown(
    user: Annotated[User, Depends(get_authorized_user)],
) -> JSONResponse

Endpoint to end the application.

Source code in packages/pyxecm/src/pyxecm_api/common/router.py

@router.get("/api/shutdown", include_in_schema=False)
def shutdown(user: Annotated[User, Depends(get_authorized_user)]) -> JSONResponse:
    """Endpoint to end the application."""

    logger.warning(
        "Shutting down the API - Requested via api by user -> %s",
        user.id,
    )
    os.kill(os.getpid(), signal.SIGTERM)

    return JSONResponse({"status": "shutdown"}, status_code=HTTPStatus.ACCEPTED)

list_browser_automation_files

list_browser_automation_files(
    user: Annotated[User, Depends(get_authorized_user)],
) -> JSONResponse

List all browser automation files.

Source code in packages/pyxecm/src/pyxecm_api/common/router.py

@router.get(path="/browser_automations/assets", tags=["payload"])
def list_browser_automation_files(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
) -> JSONResponse:
    """List all browser automation files."""

    result = list_files_in_directory(
        os.path.join(
            tempfile.gettempdir(),
            "browser_automations",
        )
    )

    return JSONResponse(result)

get_browser_automation_file

get_browser_automation_file(
    user: Annotated[User, Depends(get_authorized_user)],
    file: Annotated[str, Query(description="File name")],
) -> FileResponse

Download the logfile for a specific payload.

Source code in packages/pyxecm/src/pyxecm_api/common/router.py

@router.get(path="/browser_automations/download", tags=["payload"])
def get_browser_automation_file(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    file: Annotated[str, Query(description="File name")],
) -> FileResponse:
    """Download the logfile for a specific payload."""

    filename = os.path.join(tempfile.gettempdir(), "browser_automations", file)

    if not os.path.isfile(filename):
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="File -> '{}' not found".format(filename),
        )

    media_type, _ = mimetypes.guess_type(filename)

    with open(filename, "rb") as f:
        content = f.read()

    return Response(
        content,
        media_type=media_type,
        headers={
            "Content-Disposition": f'attachment; filename="{os.path.basename(filename)}"',
        },
    )

models

Define common base Models.

Classes:

Name	Description
CustomizerStatus	Define Model for Customizer Status.

CustomizerStatus

Bases: BaseModel

Define Model for Customizer Status.

metrics

Metrics for payload logs.

Functions:

Name	Description
payload_logs_by_payload	Metrics for payload logs by payload.
payload_logs_total	Metrics for total payload logs messages.

payload_logs_by_payload

payload_logs_by_payload(
    payload_list: PayloadList,
) -> Callable[[Info], None]

Metrics for payload logs by payload.

Source code in packages/pyxecm/src/pyxecm_api/common/metrics.py

def payload_logs_by_payload(payload_list: PayloadList) -> Callable[[Info], None]:
    """Metrics for payload logs by payload."""

    metrics_error = Gauge(
        "payload_error",
        "Number of ERROR log messages for by payload",
        labelnames=("index", "name", "logfile"),
    )

    metrics_warning = Gauge(
        "payload_warning",
        "Number of WARNING log messages for by payload",
        labelnames=("index", "name", "logfile"),
    )

    metrics_info = Gauge(
        "payload_info",
        "Number of INFO log messages for by payload",
        labelnames=("index", "name", "logfile"),
    )

    metrics_debug = Gauge(
        "payload_debug",
        "Number of DEBUG log messages for by payload",
        labelnames=("index", "name", "logfile"),
    )

    def instrumentation(info: Info) -> None:  # noqa: ARG001
        df = payload_list.get_payload_items()
        data = [{"index": idx, **row} for idx, row in df.iterrows()]

        for item in data:
            metrics_error.labels(item["index"], item["name"], item["logfile"]).set(
                item["log_error"],
            )
            metrics_warning.labels(item["index"], item["name"], item["logfile"]).set(
                item["log_warning"],
            )
            metrics_info.labels(item["index"], item["name"], item["logfile"]).set(
                item["log_info"],
            )
            metrics_debug.labels(item["index"], item["name"], item["logfile"]).set(
                item["log_debug"],
            )

    return instrumentation

payload_logs_total

payload_logs_total(
    payload_list: PayloadList,
) -> Callable[[Info], None]

Metrics for total payload logs messages.

Source code in packages/pyxecm/src/pyxecm_api/common/metrics.py

def payload_logs_total(payload_list: PayloadList) -> Callable[[Info], None]:
    """Metrics for total payload logs messages."""

    metrics_error = Gauge(
        "payload_error_total",
        "Total number of ERROR log messages",
    )

    metrics_warning = Gauge(
        "payload_warning_total",
        "Total number of WARNING log messages",
    )

    metrics_info = Gauge(
        "payload_info_total",
        "Total number of INFO log messages",
    )

    metrics_debug = Gauge(
        "payload_debug_total",
        "Total number of DEBUG log messages",
    )

    def instrumentation(info: Info) -> None:  # noqa: ARG001
        df = payload_list.get_payload_items()

        metrics_error.set(df["log_error"].sum())
        metrics_warning.set(df["log_warning"].sum())
        metrics_info.set(df["log_info"].sum())
        metrics_debug.set(df["log_debug"].sum())

    return instrumentation

functions

Define common functions.

Functions:

Name	Description
get_k8s_object	Get an instance of a K8s object.
get_otcs_object	Get an instance of a Content Server (OTCS) object.
get_otcs_object_from_otcsticket	Get an instance of a Content Server (OTCS) object.
get_settings	Get the API Settings object.
get_otcs_logs_lock	Get the Logs LOCK dict.
list_files_in_directory	Recursively list files in a directory and return a nested JSON structure with URLs.

get_k8s_object

get_k8s_object() -> K8s

Get an instance of a K8s object.

Returns:

Name	Type	Description
K8s	K8s	Return a K8s object

Source code in packages/pyxecm/src/pyxecm_api/common/functions.py

def get_k8s_object() -> K8s:
    """Get an instance of a K8s object.

    Returns:
        K8s: Return a K8s object

    """

    return K8s(logger=logger, namespace=api_settings.namespace)

get_otcs_object

get_otcs_object() -> OTCS

Get an instance of a Content Server (OTCS) object.

Returns:

Name	Type	Description
OTCS	OTCS	Return a new OTCS object.

Source code in packages/pyxecm/src/pyxecm_api/common/functions.py

def get_otcs_object() -> OTCS:
    """Get an instance of a Content Server (OTCS) object.

    Returns:
        OTCS:
            Return a new OTCS object.

    """

    settings = Settings()

    otcs = OTCS(
        protocol=settings.otcs.url_backend.scheme,
        hostname=settings.otcs.url_backend.host,
        port=settings.otcs.url_backend.port,
        public_url=str(settings.otcs.url),
        username=settings.otcs.username,
        password=settings.otcs.password.get_secret_value(),
        user_partition=settings.otcs.partition,
        resource_name=settings.otcs.resource_name,
        base_path=settings.otcs.base_path,
        support_path=settings.otcs.support_path,
        download_dir=settings.otcs.download_dir,
        feme_uri=settings.otcs.feme_uri,
        logger=logger,
    )

    # Authenticate at Content Server:
    otcs.authenticate()

    return otcs

get_otcs_object_from_otcsticket

get_otcs_object_from_otcsticket(
    otcs_ticket: Annotated[str, Depends(get_otcsticket)],
) -> OTCS

Get an instance of a Content Server (OTCS) object.

Returns:

Name	Type	Description
OTCS	OTCS	Return an OTCS object.

Source code in packages/pyxecm/src/pyxecm_api/common/functions.py

def get_otcs_object_from_otcsticket(otcs_ticket: Annotated[str, Depends(get_otcsticket)]) -> OTCS:
    """Get an instance of a Content Server (OTCS) object.

    Returns:
        OTCS:
            Return an OTCS object.

    """

    settings = Settings()

    # Create an OTCS object without defining the username and password:
    otcs = OTCS(
        # protocol=settings.otcs.url_backend.scheme,
        # hostname=settings.otcs.url_backend.host,
        # port=settings.otcs.url_backend.port,
        protocol=settings.otcs.url_frontend.scheme,
        hostname=settings.otcs.url_frontend.host,
        port=settings.otcs.url_frontend.port,
        public_url=str(settings.otcs.url),
        user_partition=settings.otcs.partition,
        resource_name=settings.otcs.resource_name,
        base_path=settings.otcs.base_path,
        support_path=settings.otcs.support_path,
        download_dir=settings.otcs.download_dir,
        feme_uri=settings.otcs.feme_uri,
        logger=logger,
    )

    # Instead set the OTCS authentication ticket directly:
    otcs._otcs_ticket = otcs_ticket  # noqa: SLF001

    return otcs

get_settings

get_settings() -> CustomizerAPISettings

Get the API Settings object.

Returns:

Name	Type	Description
CustomizerPISettings	CustomizerAPISettings	Returns the API Settings.

Source code in packages/pyxecm/src/pyxecm_api/common/functions.py

def get_settings() -> CustomizerAPISettings:
    """Get the API Settings object.

    Returns:
        CustomizerPISettings:
            Returns the API Settings.

    """

    return api_settings

get_otcs_logs_lock

get_otcs_logs_lock() -> dict

Get the Logs LOCK dict.

Returns:

Type	Description
dict	The dict with all LOCKS for the logs

Source code in packages/pyxecm/src/pyxecm_api/common/functions.py

def get_otcs_logs_lock() -> dict:
    """Get the Logs LOCK dict.

    Returns:
        The dict with all LOCKS for the logs

    """

    return LOGS_LOCK

list_files_in_directory

list_files_in_directory(directory: str) -> dict

Recursively list files in a directory and return a nested JSON structure with URLs.

Source code in packages/pyxecm/src/pyxecm_api/common/functions.py

def list_files_in_directory(directory: str) -> dict:
    """Recursively list files in a directory and return a nested JSON structure with URLs."""

    result = {}
    for root, dirs, files in os.walk(directory):
        # Sort directories and files alphabetically
        dirs.sort()
        files.sort()

        current_level = result
        path_parts = root.split(os.sep)
        relative_path = os.path.relpath(root, directory)
        for part in path_parts[len(directory.split(os.sep)) :]:
            if part not in current_level:
                current_level[part] = {}
            current_level = current_level[part]
        for file in files:
            file_path = os.path.join(relative_path, file)
            current_level[file] = file_path

    return result

Maintenance

router

Define router for v1_maintenance.

Functions:

Name	Description
status	Return status of maintenance mode.
set_maintenance_mode_options	Configure the Maintenance Mode and set options.

status async

status(
    user: Annotated[User, Depends(get_authorized_user)],
    k8s_object: Annotated[K8s, Depends(get_k8s_object)],
) -> MaintenanceModel

Return status of maintenance mode.

Parameters:

Name	Type	Description	Default
user	User	Added to enforce authentication requirement	required
k8s_object	K8s	K8s object instance of pyxecm K8s class	required

Returns:

Name	Type	Description
dict	MaintenanceModel	Details of maintenance mode.

Source code in packages/pyxecm/src/pyxecm_api/v1_maintenance/router.py

@router.get(path="")
async def status(
    user: Annotated[models.User, Depends(get_authorized_user)],  # noqa: ARG001
    k8s_object: Annotated[K8s, Depends(get_k8s_object)],
) -> MaintenanceModel:
    """Return status of maintenance mode.

    Args:
        user (models.User):
            Added to enforce authentication requirement
        k8s_object (K8s):
            K8s object instance of pyxecm K8s class


    Returns:
        dict:
            Details of maintenance mode.

    """

    return get_maintenance_mode_status(k8s_object)

set_maintenance_mode_options async

set_maintenance_mode_options(
    user: Annotated[User, Depends(get_authorized_user)],
    k8s_object: Annotated[K8s, Depends(get_k8s_object)],
    config: Annotated[MaintenanceModel, Form()],
) -> MaintenanceModel

Configure the Maintenance Mode and set options.

Parameters:

Name	Type	Description	Default
user	User	Added to enforce authentication requirement	required
k8s_object	K8s	K8s object instance of pyxecm K8s class	required
config	MaintenanceModel	instance of the Maintenance Model	required

Returns:

Name	Type	Description
dict	MaintenanceModel	description

Source code in packages/pyxecm/src/pyxecm_api/v1_maintenance/router.py

@router.post(path="")
async def set_maintenance_mode_options(
    user: Annotated[models.User, Depends(get_authorized_user)],  # noqa: ARG001
    k8s_object: Annotated[K8s, Depends(get_k8s_object)],
    config: Annotated[MaintenanceModel, Form()],
) -> MaintenanceModel:
    """Configure the Maintenance Mode and set options.

    Args:
        user (models.User):
            Added to enforce authentication requirement
        k8s_object (K8s):
            K8s object instance of pyxecm K8s class
        config (MaintenanceModel):
            instance of the Maintenance Model

    Returns:
        dict: _description_

    """
    # Enable / Disable the actual Maintenance Mode
    set_maintenance_mode_via_ingress(config.enabled, k8s_object)

    if config.title:
        maint_settings.title = config.title

    if config.text:
        maint_settings.text = config.text

    if config.footer:
        maint_settings.footer = config.footer

    return get_maintenance_mode_status(k8s_object)

models

Define Models for the Maintenance Page Config.

Classes:

Name	Description
MaintenanceModel	Status object of the Maintenance Page.

MaintenanceModel

Bases: BaseModel

Status object of the Maintenance Page.

functions

Define functions for v1_maintenance.

Functions:

Name	Description
get_cshost	Get the cs_hostname from the environment Variable OTCS_PUBLIC_HOST otherwise read it from the otcs-frontend-configmap.
get_maintenance_mode_status	Get the status of maintenance mode by inspecting the ingress object.
set_maintenance_mode_via_ingress	Set maintenance mode.

get_cshost

get_cshost(k8s_object: K8s) -> str

Get the cs_hostname from the environment Variable OTCS_PUBLIC_HOST otherwise read it from the otcs-frontend-configmap.

Source code in packages/pyxecm/src/pyxecm_api/v1_maintenance/functions.py

def get_cshost(k8s_object: K8s) -> str:
    """Get the cs_hostname from the environment Variable OTCS_PUBLIC_HOST otherwise read it from the otcs-frontend-configmap."""

    if "OTCS_PUBLIC_URL" in os.environ:
        return os.getenv("OTCS_PUBLIC_URL", "otcs")

    else:
        cm = k8s_object.get_config_map("otcs-frontend-configmap")

        if cm is None:
            raise HTTPException(
                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
                detail=f"Could not read otcs-frontend-configmap from namespace: {k8s_object.get_namespace()}",
            )

        config_file = cm.data.get("config.yaml")
        config = yaml.safe_load(config_file)

        try:
            cs_url = HttpUrl(config.get("csurl"))
        except ValidationError as ve:
            raise HTTPException(
                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
                detail="Could not read otcs_host from environment variable OTCS_PULIBC_URL or configmap otcs-frontend-configmap/config.yaml/cs_url",
            ) from ve
        return cs_url.host

get_maintenance_mode_status

get_maintenance_mode_status(k8s_object: K8s) -> dict

Get the status of maintenance mode by inspecting the ingress object.

Parameters:

Name	Type	Description	Default
k8s_object	K8s	An instance of the K8s class used to interact with Kubernetes resources.	required

Returns:

Name	Type	Description
dict	dict	A dictionary containing the details of the maintenance mode, including: - enabled (bool): Indicates whether maintenance mode is enabled. - title (str): The title of the maintenance mode message. - text (str): The text of the maintenance mode message. - footer (str): The footer of the maintenance mode message.

Raises:

Type	Description
HTTPException	If no ingress object is found to determine the maintenance mode status.

Source code in packages/pyxecm/src/pyxecm_api/v1_maintenance/functions.py

def get_maintenance_mode_status(k8s_object: K8s) -> dict:
    """Get the status of maintenance mode by inspecting the ingress object.

    Args:
        k8s_object (K8s): An instance of the K8s class used to interact with Kubernetes resources.


    Returns:
        dict: A dictionary containing the details of the maintenance mode, including:
            - enabled (bool): Indicates whether maintenance mode is enabled.
            - title (str): The title of the maintenance mode message.
            - text (str): The text of the maintenance mode message.
            - footer (str): The footer of the maintenance mode message.

    Raises:
        HTTPException: If no ingress object is found to determine the maintenance mode status.

    """
    ingress = k8s_object.get_ingress("otxecm-ingress")

    if ingress is None:
        raise HTTPException(
            status_code=500,
            detail="No ingress object found to read Maintenance Mode status",
        )

    enabled = False
    for rule in ingress.spec.rules:
        if rule.host == get_cshost(k8s_object):
            for path in rule.http.paths:
                if path.path == "/":
                    enabled = path.backend.service.name != "otcs-frontend"
                    break

    return MaintenanceModel(
        enabled=enabled, title=maint_settings.title, text=maint_settings.text, footer=maint_settings.footer
    )

set_maintenance_mode_via_ingress

set_maintenance_mode_via_ingress(
    enabled: bool, k8s_object: K8s
) -> None

Set maintenance mode.

Source code in packages/pyxecm/src/pyxecm_api/v1_maintenance/functions.py

def set_maintenance_mode_via_ingress(enabled: bool, k8s_object: K8s) -> None:
    """Set maintenance mode."""

    logger.warning(
        "Setting Maintenance Mode to -> %s",
        (enabled),
    )

    if enabled:
        k8s_object.update_ingress_backend_services(
            "otxecm-ingress",
            get_cshost(k8s_object=k8s_object),
            "customizer",
            5555,
        )
    else:
        k8s_object.update_ingress_backend_services(
            "otxecm-ingress",
            get_cshost(k8s_object=k8s_object),
            "otcs-frontend",
            80,
        )

OTCS

router

Define router for v1_otcs.

Functions:

Name	Description
put_otcs_logs	Collect the logs from the given OTCS instances.
post_otcs_log_file	Upload a file to disk.
get_otcs_log_files	List all otcs logs that can be downloaded.
delete_otcs_log_files	Delete all otcs log files.
delete_otcs_log_file	Delete single OTCS log archive.
get_otcs_log_file	Download OTCS log archive.

put_otcs_logs async

put_otcs_logs(
    user: Annotated[User, Depends(get_authorized_user)],
    k8s_object: Annotated[K8s, Depends(get_k8s_object)],
    settings: Annotated[
        CustomizerAPISettings, Depends(get_settings)
    ],
    otcs_logs_lock: Annotated[
        dict[str, Lock], Depends(get_otcs_logs_lock)
    ],
    hosts: Annotated[list[str], Body()],
) -> JSONResponse

Collect the logs from the given OTCS instances.

Source code in packages/pyxecm/src/pyxecm_api/v1_otcs/router.py

@router.put(path="/logs", tags=["otcs"])
async def put_otcs_logs(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    k8s_object: Annotated[K8s, Depends(get_k8s_object)],
    settings: Annotated[CustomizerAPISettings, Depends(get_settings)],
    otcs_logs_lock: Annotated[dict[str, Lock], Depends(get_otcs_logs_lock)],
    hosts: Annotated[list[str], Body()],
) -> JSONResponse:
    """Collect the logs from the given OTCS instances."""

    if "all" in hosts:
        hosts = []
        for sts in ["otcs-admin", "otcs-frontend", "otcs-backend-search", "otcs-da"]:
            try:
                sts_replicas = k8s_object.get_stateful_set_scale(sts)

                if sts_replicas is None:
                    logger.debug("Cannot get statefulset {sts}")
                    continue

                hosts.extend([f"{sts}-{i}" for i in range(sts_replicas.status.replicas)])
            except Exception as e:
                raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR) from e

    msg = {}
    for host in hosts:
        if host not in otcs_logs_lock:
            otcs_logs_lock[host] = Lock()

        if not otcs_logs_lock[host].locked():
            Thread(target=collect_otcs_logs, args=(host, k8s_object, otcs_logs_lock[host], settings)).start()
            msg[host] = {"status": "ok", "message": "Logs are being collected"}
        else:
            msg[host] = {"status": "error", "message": "Logs are already being collected"}

    status = (
        HTTPStatus.TOO_MANY_REQUESTS if any(msg[host]["status"] == "error" for host in msg) else HTTPStatus.ACCEPTED
    )
    return JSONResponse(msg, status_code=status)

post_otcs_log_file async

post_otcs_log_file(
    file: Annotated[UploadFile, File(...)],
    settings: Annotated[
        CustomizerAPISettings, Depends(get_settings)
    ],
    key: str = "",
) -> JSONResponse

Upload a file to disk.

Parameters:

Name	Type	Description	Default
file	Annotated[UploadFile, File(...)]	File to be uploaded.	required
settings	Annotated[CustomizerAPISettings, Depends(get_settings)]	CustomizerAPISettings.	required
key	str	Key to validate the upload.	''

Returns:

Name	Type	Description
JSONResponse	JSONResponse	Status of the upload

Source code in packages/pyxecm/src/pyxecm_api/v1_otcs/router.py

@router.post("/logs/upload", tags=["otcs"], include_in_schema=True)
async def post_otcs_log_file(
    file: Annotated[UploadFile, File(...)],
    settings: Annotated[CustomizerAPISettings, Depends(get_settings)],
    key: str = "",
) -> JSONResponse:
    """Upload a file to disk.

    Args:
        file: File to be uploaded.
        settings: CustomizerAPISettings.
        key: Key to validate the upload.

    Returns:
        JSONResponse: Status of the upload

    """
    if key != settings.upload_key:
        raise HTTPException(status_code=403, detail="Invalid Uploadkey")

    upload_dir = Path(settings.upload_folder)
    await upload_dir.mkdir(parents=True, exist_ok=True)

    try:
        file_path = str(upload_dir / file.filename)
        async with await anyio.open_file(file_path, "wb") as f:
            # Process the file in chunks instead of loading the entire file into memory
            while True:
                chunk = await file.read(65536)  # Read 64KB at a time
                if not chunk:
                    break
                await f.write(chunk)
    except Exception as e:
        raise HTTPException(status_code=500, detail="Something went wrong") from e
    finally:
        await file.close()

    return {"message": f"Successfully uploaded {file.filename}"}

get_otcs_log_files async

get_otcs_log_files(
    user: Annotated[User, Depends(get_authorized_user)],
    settings: Annotated[
        CustomizerAPISettings, Depends(get_settings)
    ],
    k8s_object: Annotated[K8s, Depends(get_k8s_object)],
    otcs_logs_lock: Annotated[
        dict[str, Lock], Depends(get_otcs_logs_lock)
    ],
) -> JSONResponse

List all otcs logs that can be downloaded.

Source code in packages/pyxecm/src/pyxecm_api/v1_otcs/router.py

@router.get("/logs", tags=["otcs"])
async def get_otcs_log_files(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    settings: Annotated[CustomizerAPISettings, Depends(get_settings)],
    k8s_object: Annotated[K8s, Depends(get_k8s_object)],
    otcs_logs_lock: Annotated[dict[str, Lock], Depends(get_otcs_logs_lock)],
) -> JSONResponse:
    """List all otcs logs that can be downloaded."""

    upload_dir = Path(settings.upload_folder)
    await upload_dir.mkdir(parents=True, exist_ok=True)

    files = []
    entries = [entry.name async for entry in upload_dir.iterdir()]

    for filename in sorted(entries):
        file_path = upload_dir / filename
        if await file_path.is_file():
            stat = await file_path.stat()
            file_size = stat.st_size
            files.append({"filename": filename, "size": file_size})

    response = {"status": {host: bool(otcs_logs_lock[host].locked()) for host in otcs_logs_lock}, "files": files}

    # Extend response with all hosts
    for sts in ["otcs-admin", "otcs-frontend", "otcs-backend-search", "otcs-da"]:
        try:
            sts_replicas = k8s_object.get_stateful_set_scale(sts)

            if sts_replicas is None:
                logger.debug("Cannot get statefulset {sts}")
                continue

            for i in range(sts_replicas.status.replicas):
                host = f"{sts}-{i}"

                if host in otcs_logs_lock:
                    response["status"][host] = otcs_logs_lock[host].locked()
                else:
                    response["status"][host] = False

        except Exception as e:
            raise HTTPException(status_code=HTTPStatus.INTERNAL_SERVER_ERROR) from e

    return JSONResponse(response, status_code=HTTPStatus.OK)

delete_otcs_log_files async

delete_otcs_log_files(
    user: Annotated[User, Depends(get_authorized_user)],
    settings: Annotated[
        CustomizerAPISettings, Depends(get_settings)
    ],
) -> JSONResponse

Delete all otcs log files.

Source code in packages/pyxecm/src/pyxecm_api/v1_otcs/router.py

@router.delete("/logs", tags=["otcs"])
async def delete_otcs_log_files(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    settings: Annotated[CustomizerAPISettings, Depends(get_settings)],
) -> JSONResponse:
    """Delete all otcs log files."""
    # Use a thread to run the blocking rmtree to avoid blocking the event loop
    await anyio.to_thread.run_sync(shutil.rmtree, settings.upload_folder)
    return JSONResponse({"message": "Successfully deleted all files"}, status_code=HTTPStatus.OK)

delete_otcs_log_file async

delete_otcs_log_file(
    user: Annotated[User, Depends(get_authorized_user)],
    settings: Annotated[
        CustomizerAPISettings, Depends(get_settings)
    ],
    file_name: str,
) -> FileResponse

Delete single OTCS log archive.

Source code in packages/pyxecm/src/pyxecm_api/v1_otcs/router.py

@router.delete("/logs/{file_name}", tags=["otcs"])
async def delete_otcs_log_file(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    settings: Annotated[CustomizerAPISettings, Depends(get_settings)],
    file_name: str,
) -> FileResponse:
    """Delete single OTCS log archive."""
    upload_dir = Path(settings.upload_folder)
    file_path = upload_dir / file_name

    if not await file_path.exists():
        raise HTTPException(status_code=404, detail="File not found")

    try:
        await anyio.to_thread.run_sync(os.remove, str(file_path))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"{e}") from e

    return JSONResponse({"message": f"Successfully deleted {file_name}"}, status_code=HTTPStatus.OK)

get_otcs_log_file async

get_otcs_log_file(
    user: Annotated[User, Depends(get_authorized_user)],
    settings: Annotated[
        CustomizerAPISettings, Depends(get_settings)
    ],
    file_name: str,
) -> FileResponse

Download OTCS log archive.

Source code in packages/pyxecm/src/pyxecm_api/v1_otcs/router.py

@router.get("/logs/{file_name}", tags=["otcs"])
async def get_otcs_log_file(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    settings: Annotated[CustomizerAPISettings, Depends(get_settings)],
    file_name: str,
) -> FileResponse:
    """Download OTCS log archive."""
    upload_dir = Path(settings.upload_folder)
    file_path = upload_dir / file_name
    if not await file_path.exists():
        raise HTTPException(status_code=404, detail="File not found")

    return FileResponse(str(file_path), media_type="application/octet-stream", filename=file_name)

functions

Define functions for v1_otcs.

Functions:

Name	Description
collect_otcs_logs	Collect the logs for the given OTCS instance.

collect_otcs_logs

collect_otcs_logs(
    host: str,
    k8s_object: K8s,
    logs_lock: Lock,
    settings: CustomizerAPISettings,
) -> None

Collect the logs for the given OTCS instance.

Source code in packages/pyxecm/src/pyxecm_api/v1_otcs/functions.py

def collect_otcs_logs(host: str, k8s_object: K8s, logs_lock: Lock, settings: CustomizerAPISettings) -> None:
    """Collect the logs for the given OTCS instance."""

    with logs_lock:
        timestamp = datetime.now(tz=UTC).strftime("%Y-%m-%d_%H-%M")
        tgz_file = f"/tmp/{timestamp}_{host}.tar.gz"  # noqa: S108

        if host.startswith("otcs-frontend"):
            container = "otcs-frontend-container"
        elif host.startswith("otcs-backend-search"):
            container = "otcs-backend-search-container"
        elif host.startswith("otcs-admin"):
            container = "otcs-admin-container"
        elif host.startswith("otcs-da"):
            container = "otcs-da-container"
        else:
            container = None

        logger.info("Collecting logs for %s", host)
        k8s_object.exec_pod_command(
            pod_name=host,
            command=["tar", "-czvf", tgz_file, "/opt/opentext/cs/logs", "/opt/opentext/cs_persist/contentserver.log"],
            container=container,
            timeout=1800,
        )

        logger.info("Uploading logs for %s", host)
        k8s_object.exec_pod_command(
            pod_name=host,
            command=[
                "curl",
                "-X",
                "POST",
                "-F",
                f"file=@{tgz_file}",
                f"{settings.upload_url}?key={settings.upload_key}",
            ],
            container=container,
            timeout=1800,
        )

        logger.info("Cleanup logs for %s", host)
        k8s_object.exec_pod_command(
            pod_name=host,
            command=["rm", tgz_file],
            container=container,
        )

Payload

router

Define router for v1_payload.

Functions:

Name	Description
create_payload_item	Upload a new payload item.
get_payload_items	Get all Payload items.
get_payload_item	Get a payload item based on its ID.
update_payload_item	Update an existing payload item.
delete_payload_item	Delete an existing payload item.
move_payload_item_up	Move a payload item up in the list.
move_payload_item_down	Move a payload item down in the list.
get_payload_content	Get a payload item based on its ID.
download_payload_content	Download the payload for a specific payload item.
download_payload_logfile	Download the logfile for a specific payload.
get_payload_logfile	Get the log file of a payload for the endpoints that read parts of it.
get_payload_logfile_info	Describe the logfile of a payload and the progress of its line index.
get_payload_logfile_lines	Read a window of lines from the logfile of a payload, optionally filtered.
stream_logfile	Stream the logfile and follow changes.

create_payload_item async

create_payload_item(
    user: Annotated[User, Depends(get_authorized_user)],
    settings: Annotated[
        CustomizerAPISettings, Depends(get_settings)
    ],
    upload_file: Annotated[UploadFile, File(...)],
    name: Annotated[str, Form()] = "",
    dependencies: Annotated[
        list[int] | list[str] | None, Form()
    ] = None,
    enabled: Annotated[bool, Form()] = True,
    loglevel: Annotated[
        Literal["DEBUG", "INFO", "WARNING"] | None,
        Form(
            description="Loglevel for the Payload processing"
        ),
    ] = "INFO",
) -> PayloadListItem

Upload a new payload item.

Parameters:

Name	Type	Description	Default
user	User	The user who is uploading the payload. Defaults to None.	required
settings	CustomizerAPISettings	The settings object.	required
upload_file	UploadFile	The file to upload. Defaults to File(...).	required
name	str	The name of the payload (if not provided we will use the file name).	''
dependencies	list of integers	List of other payload items this item depends on.	None
enabled	bool	Flag indicating if the payload is enabled or not.	True
loglevel	str	The loglevel for the payload processing. Defaults to "INFO".	'INFO'

Raises:

Type	Description
HTTPException	Raised, if payload list is not initialized.

Returns:

Name	Type	Description
dict	PayloadListItem	The HTTP response.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.post(path="")
async def create_payload_item(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    settings: Annotated[CustomizerAPISettings, Depends(get_settings)],
    upload_file: Annotated[UploadFile, File(...)],
    name: Annotated[str, Form()] = "",
    dependencies: Annotated[list[int] | list[str] | None, Form()] = None,
    enabled: Annotated[bool, Form()] = True,
    loglevel: Annotated[
        Literal["DEBUG", "INFO", "WARNING"] | None,
        Form(
            description="Loglevel for the Payload processing",
        ),
    ] = "INFO",
) -> PayloadListItem:
    """Upload a new payload item.

    Args:
        user (User, optional):
            The user who is uploading the payload. Defaults to None.
        settings (CustomizerAPISettings):
            The settings object.
        upload_file (UploadFile, optional):
            The file to upload. Defaults to File(...).
        name (str, optional):
            The name of the payload (if not provided we will use the file name).
        dependencies (list of integers):
            List of other payload items this item depends on.
        enabled (bool):
            Flag indicating if the payload is enabled or not.
        loglevel (str, optional):
            The loglevel for the payload processing. Defaults to "INFO".

    Raises:
        HTTPException:
            Raised, if payload list is not initialized.

    Returns:
        dict:
            The HTTP response.

    """
    if dependencies:
        dependencies = prepare_dependencies(dependencies)

    # Set name if not provided
    name = name or os.path.splitext(os.path.basename(upload_file.filename))[0]
    file_extension = os.path.splitext(upload_file.filename)[1]
    file_name = os.path.join(settings.temp_dir, f"{name}{file_extension}")

    # Read upload file asynchronously and write to disk using anyio
    content = await upload_file.read()
    async with await anyio.open_file(file_name, "wb") as buffer:
        await buffer.write(content)

    if dependencies == [-1]:
        dependencies = []

    return PayloadListItem(
        PAYLOAD_LIST.add_payload_item(
            name=name,
            filename=file_name,
            status="planned",
            logfile=os.path.join(settings.temp_dir, "{}.log".format(name)),
            dependencies=dependencies or [],
            enabled=enabled,
            loglevel=loglevel,
        )
    )

get_payload_items async

get_payload_items(
    user: Annotated[User, Depends(get_authorized_user)],
) -> PayloadListItems

Get all Payload items.

Raises:

Type	Description
HTTPException	payload list not initialized
HTTPException	payload list is empty

Returns:

Name	Type	Description
dict	PayloadListItems	HTTP response with the result data

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.get(path="")
async def get_payload_items(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
) -> PayloadListItems:
    """Get all Payload items.

    Raises:
        HTTPException: payload list not initialized
        HTTPException: payload list is empty

    Returns:
        dict:
            HTTP response with the result data

    """

    df = PAYLOAD_LIST.get_payload_items()

    if df is None:
        raise HTTPException(
            status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
            detail="Payload list is empty.",
        )

    data = [PayloadListItem(index=idx, **row) for idx, row in df.iterrows()]

    stats = {
        "count": len(df),
        "status": df["status"].value_counts().to_dict(),
        "logs": {
            "debug": df["log_debug"].sum(),
            "info": df["log_info"].sum(),
            "warning": df["log_warning"].sum(),
            "error": df["log_error"].sum(),
            "critical": df["log_critical"].sum(),
        },
    }

    return PayloadListItems(stats=stats, results=data)

get_payload_item async

get_payload_item(
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
) -> PayloadListItem

Get a payload item based on its ID.

Parameters:

Name	Type	Description	Default
user	Annotated[User, Depends(get_authorized_user)]	Annotated[User, Depends(get_authorized_user)]	required
payload_id	int	payload item ID	required

Raises:

Type	Description
HTTPException	a payload item with the given ID couldn't be found

Returns:

Name	Type	Description
dict	PayloadListItem	HTTP response.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.get(path="/{payload_id}")
async def get_payload_item(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    payload_id: int,
) -> PayloadListItem:
    """Get a payload item based on its ID.

    Args:
        user: Annotated[User, Depends(get_authorized_user)]
        payload_id (int): payload item ID

    Raises:
        HTTPException: a payload item with the given ID couldn't be found

    Returns:
        dict:
            HTTP response.

    """
    data = PAYLOAD_LIST.get_payload_item(index=payload_id)

    if data is None:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Payload with index -> {} not found".format(payload_id),
        )

    return PayloadListItem(index=payload_id, **data, asd="123")

update_payload_item async

update_payload_item(
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
    name: Annotated[str | None, Form()] = None,
    dependencies: Annotated[
        list[int] | list[str] | None, Form()
    ] = None,
    enabled: Annotated[bool | None, Form()] = None,
    status: Annotated[
        Literal["planned", "completed"] | None, Form()
    ] = None,
    loglevel: Annotated[
        Literal["DEBUG", "INFO", "WARNING"] | None,
        Form(
            description="Loglevel for the Payload processing"
        ),
    ] = None,
    customizer_settings: Annotated[
        str | None, Form()
    ] = None,
) -> UpdatedPayloadListItem

Update an existing payload item.

Parameters:

Name	Type	Description	Default
user	Optional[User]	User performing the update.	required
payload_id	int	ID of the payload to update.	required
name	Optional[str]	Updated name.	None
dependencies	Optional[List[int]]	Updated list of dependencies.	None
enabled	Optional[bool]	Updated enabled status.	None
loglevel	Optional[str]	Updated loglevel.	None
status	Optional[str]	Updated status.	None
customizer_settings	Optional[str]	Updated customizer settings.	None

Returns:

Name	Type	Description
dict	UpdatedPayloadListItem	HTTP response with the updated payload details.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.put(path="/{payload_id}")
async def update_payload_item(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    payload_id: int,
    name: Annotated[str | None, Form()] = None,
    dependencies: Annotated[list[int] | list[str] | None, Form()] = None,
    enabled: Annotated[bool | None, Form()] = None,
    status: Annotated[Literal["planned", "completed"] | None, Form()] = None,
    loglevel: Annotated[
        Literal["DEBUG", "INFO", "WARNING"] | None,
        Form(
            description="Loglevel for the Payload processing",
        ),
    ] = None,
    customizer_settings: Annotated[str | None, Form()] = None,
) -> UpdatedPayloadListItem:
    """Update an existing payload item.

    Args:
        user (Optional[User]): User performing the update.
        payload_id (int): ID of the payload to update.
        name (Optional[str]): Updated name.
        dependencies (Optional[List[int]]): Updated list of dependencies.
        enabled (Optional[bool]): Updated enabled status.
        loglevel (Optional[str]): Updated loglevel.
        status (Optional[str]): Updated status.
        customizer_settings (Optional[str]): Updated customizer settings.

    Returns:
        dict: HTTP response with the updated payload details.

    """

    if dependencies:
        dependencies = prepare_dependencies(dependencies)
    # Check if the payload exists
    payload_item = PAYLOAD_LIST.get_payload_item(
        payload_id,
    )  # Assumes a method to retrieve payload by ID
    if payload_item is None:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Payload with ID -> {} not found.".format(payload_id),
        )

    update_data = {}

    # Update fields if provided
    if name is not None:
        update_data["name"] = name
    if dependencies is not None:
        update_data["dependencies"] = dependencies
    if enabled is not None:
        update_data["enabled"] = enabled
    if status is not None:
        update_data["status"] = status
    if loglevel is not None:
        update_data["loglevel"] = loglevel

        thread_logger = logging.getLogger(name=f"Payload_{payload_id}")
        thread_logger.setLevel(loglevel)

    if customizer_settings is not None:
        try:
            update_data["customizer_settings"] = json.loads(customizer_settings)
        except Exception as e:
            raise HTTPException(detail=e, status_code=HTTPStatus.BAD_REQUEST) from e

    if "status" in update_data and update_data["status"] == "planned":
        logger.info("Resetting log message counters for -> %s", payload_id)
        update_data["log_debug"] = 0
        update_data["log_info"] = 0
        update_data["log_warning"] = 0
        update_data["log_error"] = 0
        update_data["log_critical"] = 0

        update_data["start_time"] = None
        update_data["stop_time"] = None
        update_data["duration"] = None

        data = PAYLOAD_LIST.get_payload_item(index=payload_id)
        if await anyio.to_thread.run_sync(os.path.isfile, data.logfile):
            logger.info(
                "Deleting log file (for payload) -> %s (%s)",
                data.logfile,
                payload_id,
            )

            now = datetime.now(UTC)
            old_log_name = (
                os.path.dirname(data.logfile)
                + "/"
                + os.path.splitext(os.path.basename(data.logfile))[0]
                + now.strftime("_%Y-%m-%d_%H-%M-%S.log")
            )

            await anyio.to_thread.run_sync(os.rename, data.logfile, old_log_name)

    # Save the updated payload back to the list (or database)
    result = PAYLOAD_LIST.update_payload_item(
        index=payload_id,
        update_data=update_data,
    )  # Assumes a method to update the payload
    if not result:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Failed to update Payload with ID -> {} with data -> {}".format(
                payload_id,
                update_data,
            ),
        )

    return UpdatedPayloadListItem(
        message="Payload updated successfully",
        payload=PayloadListItem(index=payload_id, **PAYLOAD_LIST.get_payload_item(index=payload_id)),
        updated_fields=update_data,
    )

delete_payload_item async

delete_payload_item(
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
) -> JSONResponse

Delete an existing payload item.

Parameters:

Name	Type	Description	Default
user	Optional[User]	User performing the update.	required
payload_id	int	The ID of the payload to update.	required

Returns:

Name	Type	Description
dict	JSONResponse	response or None

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.delete(path="/{payload_id}")
async def delete_payload_item(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    payload_id: int,
) -> JSONResponse:
    """Delete an existing payload item.

    Args:
        user (Optional[User]): User performing the update.
        payload_id (int): The ID of the payload to update.

    Returns:
        dict: response or None

    """

    # Check if the payload exists
    result = PAYLOAD_LIST.remove_payload_item(payload_id)
    if not result:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Payload with ID -> {} not found.".format(payload_id),
        )

move_payload_item_up async

move_payload_item_up(
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
) -> dict

Move a payload item up in the list.

Parameters:

Name	Type	Description	Default
user	Annotated[User, Depends(get_authorized_user)]	Annotated[User, Depends(get_authorized_user)]	required
payload_id	int	payload item ID	required

Raises:

Type	Description
HTTPException	a payload item with the given ID couldn't be found

Returns:

Name	Type	Description
dict	dict	HTTP response

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.put(path="/{payload_id}/up")
async def move_payload_item_up(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    payload_id: int,
) -> dict:
    """Move a payload item up in the list.

    Args:
        user: Annotated[User, Depends(get_authorized_user)]
        payload_id (int): payload item ID

    Raises:
        HTTPException: a payload item with the given ID couldn't be found

    Returns:
        dict: HTTP response

    """

    position = PAYLOAD_LIST.move_payload_item_up(index=payload_id)

    if position is None:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Payload item with index -> {} is either out of range or is already on top of the payload list!".format(
                payload_id,
            ),
        )

    return {"result": {"new_position": position}}

move_payload_item_down async

move_payload_item_down(
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
) -> dict

Move a payload item down in the list.

Parameters:

Name	Type	Description	Default
user	Annotated[User, Depends(get_authorized_user)]	Annotated[User, Depends(get_authorized_user)]	required
payload_id	int	The payload item ID.	required

Raises:

Type	Description
HTTPException	a payload item with the given ID couldn't be found

Returns:

Name	Type	Description
dict	dict	HTTP response

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.put(path="/{payload_id}/down")
async def move_payload_item_down(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    payload_id: int,
) -> dict:
    """Move a payload item down in the list.

    Args:
        user: Annotated[User, Depends(get_authorized_user)]
        payload_id (int):
            The payload item ID.

    Raises:
        HTTPException: a payload item with the given ID couldn't be found

    Returns:
        dict: HTTP response

    """

    position = PAYLOAD_LIST.move_payload_item_down(index=payload_id)

    if position is None:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Payload item with index -> {} is either out of range or is already on bottom of the payload list!".format(
                payload_id,
            ),
        )

    return {"result": {"new_position": position}}

get_payload_content async

get_payload_content(
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
) -> dict | None

Get a payload item based on its ID.

Parameters:

Name	Type	Description	Default
user	Annotated[User, Depends(get_authorized_user)]	Annotated[User, Depends(get_authorized_user)]	required
payload_id	int	The payload item ID.	required

Raises:

Type	Description
HTTPException	A payload item with the given ID couldn't be found.

Returns:

Name	Type	Description
dict	dict | None	HTTP response.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.get(path="/{payload_id}/content")
async def get_payload_content(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    # pylint: disable=unused-argument
    payload_id: int,
) -> dict | None:
    """Get a payload item based on its ID.

    Args:
        user: Annotated[User, Depends(get_authorized_user)]
        payload_id (int):
            The payload item ID.

    Raises:
        HTTPException:
            A payload item with the given ID couldn't be found.

    Returns:
        dict:
            HTTP response.

    """

    data = PAYLOAD_LIST.get_payload_item(index=payload_id)

    if data is None:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Payload with ID -> {} not found!".format(payload_id),
        )

    filename = data.filename

    return load_payload(payload_source=filename)

download_payload_content async

download_payload_content(
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
) -> FileResponse

Download the payload for a specific payload item.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.get(path="/{payload_id}/download")
async def download_payload_content(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    payload_id: int,
) -> FileResponse:
    """Download the payload for a specific payload item."""

    payload = PAYLOAD_LIST.get_payload_item(index=payload_id)

    if payload is None:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Payload with ID -> {} not found!".format(payload_id),
        )

    if not await anyio.to_thread.run_sync(os.path.isfile, payload.filename):
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Payload file -> '{}' not found".format(payload.filename),
        )

    async with await anyio.open_file(payload.filename, "rb") as file:
        content = await file.read()

    if payload.filename.endswith(".gz.b64"):
        content = base64.b64decode(content)
        content = gzip.decompress(content)

    download_name = (
        os.path.basename(payload.filename.removesuffix(".gz.b64"))
        if payload.filename.endswith(".gz.b64")
        else os.path.basename(payload.filename)
    )

    return Response(
        content,
        media_type="application/octet-stream",
        headers={
            "Content-Disposition": f'attachment; filename="{download_name}"',
        },
    )

download_payload_logfile async

download_payload_logfile(
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
) -> FileResponse

Download the logfile for a specific payload.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.get(path="/{payload_id}/log")
async def download_payload_logfile(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    payload_id: int,
) -> FileResponse:
    """Download the logfile for a specific payload."""

    payload = PAYLOAD_LIST.get_payload_item(index=payload_id)

    if payload is None:
        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Payload not found")

    filename = payload.logfile

    if not await anyio.to_thread.run_sync(os.path.isfile, filename):
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Log file -> '{}' not found".format(filename),
        )

    # The file is streamed from disk (a log can have several gigabytes) and a part of it can be
    # requested with a "Range" header. Without one the complete file is sent.
    return GrowingFileResponse(
        filename,
        media_type="application/octet-stream",
        headers={
            "Content-Disposition": f'attachment; filename="{os.path.basename(filename)}"',
            # The response carries validators now (ETag, Last-Modified) but a log changes
            # all the time: keep browsers from reusing a stored copy on their own.
            "Cache-Control": "no-cache",
        },
    )

get_payload_logfile

get_payload_logfile(payload_id: int) -> tuple[str, bool]

Get the log file of a payload for the endpoints that read parts of it.

Parameters:

Name	Type	Description	Default
payload_id	int	The payload item ID.	required

Raises:

Type	Description
HTTPException	A payload item with the given ID couldn't be found.

Returns:

Type	Description
tuple[str, bool]	tuple[str, bool]: The path of the log file and whether the payload is running, i.e. whether the log is still being written.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

def get_payload_logfile(payload_id: int) -> tuple[str, bool]:
    """Get the log file of a payload for the endpoints that read parts of it.

    Args:
        payload_id (int):
            The payload item ID.

    Raises:
        HTTPException:
            A payload item with the given ID couldn't be found.

    Returns:
        tuple[str, bool]:
            The path of the log file and whether the payload is running, i.e. whether
            the log is still being written.

    """

    payload = PAYLOAD_LIST.get_payload_item(index=payload_id)

    if payload is None:
        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Payload not found")

    return payload.logfile, payload.status == "running"

get_payload_logfile_info async

get_payload_logfile_info(
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
) -> PayloadLogInfo

Describe the logfile of a payload and the progress of its line index.

The line index makes it possible to read any part of a log of any size (see /log/lines). It is built in the background when a log is looked at for the first time.

Parameters:

Name	Type	Description	Default
user	Annotated[User, Depends(get_authorized_user)]	Annotated[User, Depends(get_authorized_user)]	required
payload_id	int	The payload item ID.	required

Raises:

Type	Description
HTTPException	The payload item or its logfile couldn't be found.

Returns:

Name	Type	Description
PayloadLogInfo	PayloadLogInfo	Size, number of lines and number of log records per level of the logfile, how far the line index reaches and whether the payload is running.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.get(path="/{payload_id}/log/info")
async def get_payload_logfile_info(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    payload_id: int,
) -> PayloadLogInfo:
    """Describe the logfile of a payload and the progress of its line index.

    The line index makes it possible to read any part of a log of any size (see
    `/log/lines`). It is built in the background when a log is looked at for the first time.

    Args:
        user: Annotated[User, Depends(get_authorized_user)]
        payload_id (int):
            The payload item ID.

    Raises:
        HTTPException:
            The payload item or its logfile couldn't be found.

    Returns:
        PayloadLogInfo:
            Size, number of lines and number of log records per level of the logfile, how
            far the line index reaches and whether the payload is running.

    """

    filename, running = get_payload_logfile(payload_id)

    try:
        info = await run_log_read(None, read_log_info, filename, running)
    except FileNotFoundError as exc:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Log file -> '{}' not found".format(filename),
        ) from exc

    return PayloadLogInfo(**info)

get_payload_logfile_lines async

get_payload_logfile_lines(
    request: Request,
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
    lines_request: Annotated[
        LogLinesRequest, Depends(lines_request_from_query)
    ],
) -> PayloadLogLines

Read a window of lines from the logfile of a payload, optionally filtered.

The logfile is never read into memory as a whole, so this works for logs of any size. A request stops after limit lines, after max_bytes, at the end (or the beginning) of the logfile, or when it has worked for about a second. The response tells which byte range of the logfile it covers: continue with offset=<end>&direction=forward or with offset=<start>&direction=backward until eof or bof is true.

Parameters:

Name	Type	Description	Default
request	Request	The request. The scan stops when its client disconnects.	required
user	Annotated[User, Depends(get_authorized_user)]	Annotated[User, Depends(get_authorized_user)]	required
payload_id	int	The payload item ID.	required
lines_request	LogLinesRequest	What to read: the query parameters offset, line, direction, limit, max_bytes, levels, q, regex, case, word and count_only (see lines_request_from_query).	required

Raises:

Type	Description
HTTPException	The payload item or its logfile couldn't be found (404), or a parameter is invalid (400).

Returns:

Name	Type	Description
PayloadLogLines	PayloadLogLines	The lines (ascending), the byte range they were taken from and the state of the logfile.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.get(path="/{payload_id}/log/lines")
async def get_payload_logfile_lines(
    request: Request,
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    payload_id: int,
    lines_request: Annotated[LogLinesRequest, Depends(lines_request_from_query)],
) -> PayloadLogLines:
    """Read a window of lines from the logfile of a payload, optionally filtered.

    The logfile is never read into memory as a whole, so this works for logs of any size.
    A request stops after `limit` lines, after `max_bytes`, at the end (or the beginning) of
    the logfile, or when it has worked for about a second. The response tells which byte range
    of the logfile it covers: continue with `offset=<end>&direction=forward` or with
    `offset=<start>&direction=backward` until `eof` or `bof` is true.

    Args:
        request (Request):
            The request. The scan stops when its client disconnects.
        user: Annotated[User, Depends(get_authorized_user)]
        payload_id (int):
            The payload item ID.
        lines_request (LogLinesRequest):
            What to read: the query parameters offset, line, direction, limit, max_bytes,
            levels, q, regex, case, word and count_only (see `lines_request_from_query`).

    Raises:
        HTTPException:
            The payload item or its logfile couldn't be found (404), or a parameter
            is invalid (400).

    Returns:
        PayloadLogLines:
            The lines (ascending), the byte range they were taken from and the state of
            the logfile.

    """

    filename, running = get_payload_logfile(payload_id)

    try:
        result = await run_log_read(request, read_log_lines, filename, running, lines_request)
    except FileNotFoundError as exc:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND,
            detail="Log file -> '{}' not found".format(filename),
        ) from exc

    return PayloadLogLines(**result)

stream_logfile async

stream_logfile(
    user: Annotated[User, Depends(get_authorized_user)],
    payload_id: int,
) -> StreamingResponse

Stream the logfile and follow changes.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/router.py

@router.get(path="/{payload_id}/log/stream")
async def stream_logfile(
    user: Annotated[User, Depends(get_authorized_user)],  # noqa: ARG001
    payload_id: int,
) -> StreamingResponse:
    """Stream the logfile and follow changes."""

    payload = PAYLOAD_LIST.get_payload_item(index=payload_id)

    if payload is None:
        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Payload not found")

    filename = payload.logfile
    if await anyio.to_thread.run_sync(os.path.isfile, filename):
        return StreamingResponse(tail_log(filename), media_type="text/plain")

    raise HTTPException(status_code=HTTPStatus.NOT_FOUND)

models

Define Models for Payload.

Classes:

Name	Description
PayloadStats	Defines PayloadStats Model.
PayloadListItem	Defines PayloadListItem Model.
PayloadListItems	Defines PayloadListItems Model.
UpdatedPayloadListItem	Defines UpdatedPayloadListItem Model.
PayloadLogInfo	Defines PayloadLogInfo Model: a payload log file and the progress of its line index.
PayloadLogLine	Defines PayloadLogLine Model: one line of a payload log file.
PayloadLogLines	Defines PayloadLogLines Model: a window of lines of a payload log file.

PayloadStats

Bases: BaseModel

Defines PayloadStats Model.

PayloadListItem

Bases: BaseModel

Defines PayloadListItem Model.

PayloadListItems

Bases: BaseModel

Defines PayloadListItems Model.

UpdatedPayloadListItem

Bases: BaseModel

Defines UpdatedPayloadListItem Model.

PayloadLogInfo

Bases: BaseModel

Defines PayloadLogInfo Model: a payload log file and the progress of its line index.

PayloadLogLine

Bases: BaseModel

Defines PayloadLogLine Model: one line of a payload log file.

PayloadLogLines

Bases: BaseModel

Defines PayloadLogLines Model: a window of lines of a payload log file.

functions

API Implementation for the Customizer to start and control the payload processing.

Classes:

Name	Description
GrowingFileResponse	A file download for a file that may be appended to while it is sent.

Functions:

Name	Description
import_payload	Automatically load payload items from disk of a given directory.
prepare_dependencies	Convert the dependencies string to a list of integers.
tail_log	Asynchronously follow the log file like tail -f.
run_log_read	Run a function that reads a log file in a worker thread, off the event loop.
lines_request_from_query	Read the query parameters of a request for log lines (a dependency of the /lines endpoints).

GrowingFileResponse

Bases: FileResponse

A file download for a file that may be appended to while it is sent.

The size of the file is announced before its content is sent. The plain FileResponse then sends the file up to its end, which is more than announced if the file grew meanwhile, and the server aborts such a response. This response sends exactly the announced bytes. Everything else, Range requests included, is the behaviour of FileResponse.

import_payload

import_payload(
    payload: str | None = None,
    payload_dir: str | None = None,
    enabled: bool | None = None,
    dependencies: bool | None = None,
) -> None

Automatically load payload items from disk of a given directory.

Parameters:

Name	Type	Description	Default
payload	str	The name of the payload.	None
payload_dir	str	The local path.	None
enabled	bool	Automatically start the processing (True), or only define items (False). Defaults to False.	None
dependencies	bool	Automatically add dependency on the last payload in the queue	None

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/functions.py

def import_payload(
    payload: str | None = None,
    payload_dir: str | None = None,
    enabled: bool | None = None,
    dependencies: bool | None = None,
) -> None:
    """Automatically load payload items from disk of a given directory.

    Args:
        payload (str):
            The name of the payload.
        payload_dir (str):
            The local path.
        enabled (bool, optional):
            Automatically start the processing (True), or only define items (False).
            Defaults to False.
        dependencies (bool, optional):
            Automatically add dependency on the last payload in the queue

    """

    def import_payload_file(
        filename: str,
        enabled: bool | None,
        dependencies: bool | None,
    ) -> None:
        if not os.path.isfile(filename):
            return

        if not (filename.endswith((".yaml", ".tfvars", ".tf", ".yml.gz.b64"))):
            logger.debug("Skipping file: %s", filename)
            return

        # Load payload file
        payload_content = load_payload(filename)
        if payload_content is None:
            exception = f"The import of payload -> {filename} failed. Payload content could not be loaded."
            raise PayloadImportError(exception)

        payload_options = payload_content.get("payloadOptions", {})

        if enabled is None:
            enabled = payload_options.get("enabled", True)
            enabled = False if isinstance(enabled, str) and enabled.lower() == "false" else bool(enabled)

        # read name from options section if specified, otherwise take filename
        name = payload_options.get("name", os.path.basename(filename))

        # Get the loglevel from payloadOptions if set, otherwise use the default loglevel
        loglevel = payload_options.get("loglevel", api_settings.loglevel)

        # Get the git_url
        git_url = payload_options.get("git_url", None)

        # Dependency Management
        if dependencies is None:
            dependencies = []

            # Get all dependencies from payloadOptions and resolve their ID
            for dependency_name in payload_options.get("dependencies", []):
                dependend_item = PAYLOAD_LIST.get_payload_item_by_name(dependency_name)

                if dependend_item is None:
                    exception = (
                        f"The import of payload -> {name} failed. Dependencies cannot be resolved: {dependency_name}",
                    )
                    raise PayloadImportError(
                        exception,
                    )
                # Add the ID to the list of dependencies
                dependencies.append(dependend_item["index"])

        elif dependencies:
            try:
                payload_items = len(PAYLOAD_LIST.get_payload_items()) - 1
                dependencies = [payload_items] if payload_items != -1 else []
            except Exception:
                dependencies = []
        else:
            dependencies = []

        customizer_settings = payload_content.get("customizerSettings", {})

        logger.info("Adding payload: %s", filename)
        payload = PAYLOAD_LIST.add_payload_item(
            name=name,
            filename=filename,
            status="planned",
            logfile=f"{api_settings.logfolder}/{name}.log",
            dependencies=dependencies,
            enabled=enabled,
            git_url=git_url,
            loglevel=loglevel,
            customizer_settings=customizer_settings,
        )
        dependencies = payload["index"]

        return

    if payload is None and payload_dir is None:
        exception = "No payload or payload_dir provided"
        raise ValueError(exception)

    if payload and os.path.isdir(payload) and payload_dir is None:
        payload_dir = payload

    if payload_dir is None:
        try:
            import_payload_file(payload, enabled, dependencies)
        except PayloadImportError as exc:
            logger.error(exc)
            logger.debug(exc, exc_info=True)
        return

    elif not os.path.isdir(payload_dir):
        return

    for filename in sorted(os.listdir(payload_dir)):
        try:
            with tracer.start_as_current_span("import_payload") as t:
                t.set_attribute("payload", filename)
                import_payload_file(os.path.join(payload_dir, filename), enabled, dependencies)
        except PayloadImportError as exc:
            logger.error(exc)
            logger.debug(exc, exc_info=True)

prepare_dependencies

prepare_dependencies(dependencies: list) -> list | None

Convert the dependencies string to a list of integers.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/functions.py

def prepare_dependencies(dependencies: list) -> list | None:
    """Convert the dependencies string to a list of integers."""
    try:
        list_all = dependencies[0].split(",")
    except IndexError:
        return None

    # Remove empty values from the list
    items = list(filter(None, list_all))
    converted_list = []
    for item in items:
        try:
            converted_list.append(int(item))
        except ValueError:
            continue

    return converted_list

tail_log async

tail_log(file_path: str) -> AsyncGenerator[str]

Asynchronously follow the log file like tail -f.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/functions.py

async def tail_log(file_path: str) -> AsyncGenerator[str]:
    """Asynchronously follow the log file like `tail -f`."""
    try:
        async with await anyio.open_file(file_path) as file:
            # Move the pointer to the end of the file
            await file.seek(0, os.SEEK_END)

            while True:
                # Read new line
                line = await file.readline()
                if not line:
                    # Sleep for a little while before checking for new lines
                    await asyncio.sleep(0.5)
                    continue
                yield line
    except asyncio.exceptions.CancelledError:
        pass

run_log_read async

run_log_read(
    request: Request | None,
    function: Callable[..., Any],
    *args: Any,
) -> Any

Run a function that reads a log file in a worker thread, off the event loop.

Parameters:

Name	Type	Description	Default
request	Request | None	The request the result is for. If given, the function gets a threading.Event as its last argument, which is set as soon as the client disconnects, so that a long scan stops when nobody waits for its result any more.	required
function	Callable	The function to run.	required
*args	Any	The arguments of the function.	()

Returns:

Name	Type	Description
Any	Any	The result of the function.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/functions.py

async def run_log_read(request: Request | None, function: Callable[..., Any], *args: Any) -> Any:  # noqa: ANN401
    """Run a function that reads a log file in a worker thread, off the event loop.

    Args:
        request (Request | None):
            The request the result is for. If given, the function gets a `threading.Event`
            as its last argument, which is set as soon as the client disconnects, so that
            a long scan stops when nobody waits for its result any more.
        function (Callable):
            The function to run.
        *args (Any):
            The arguments of the function.

    Returns:
        Any:
            The result of the function.

    """

    limiter = _get_log_read_limiter()
    if request is None:
        return await anyio.to_thread.run_sync(function, *args, limiter=limiter)

    cancel = threading.Event()
    result = None
    error: Exception | None = None

    async def watch_client() -> None:
        # Most reads are done long before the first look at the connection.
        while True:
            await anyio.sleep(CLIENT_CHECK_SECONDS)
            if await request.is_disconnected():
                break
        cancel.set()

    try:
        async with anyio.create_task_group() as task_group:
            task_group.start_soon(watch_client)
            # An error must not leave the task group: it would arrive wrapped in a group.
            try:
                result = await anyio.to_thread.run_sync(function, *args, cancel, limiter=limiter)
            except Exception as exc:
                error = exc
            task_group.cancel_scope.cancel()
    finally:
        # Also stops the scan if this request itself is cancelled.
        cancel.set()

    if error is not None:
        raise error
    return result

lines_request_from_query

lines_request_from_query(
    offset: Annotated[
        int | None,
        Query(
            description="Byte offset to start at. It is snapped back to the start of the line containing it."
        ),
    ] = None,
    line: Annotated[
        int | None,
        Query(
            description="Line number (1-based) to start at. Alternative to the offset."
        ),
    ] = None,
    direction: Annotated[
        str,
        Query(
            description="'forward' returns the lines from the start position on, 'backward' the lines before it. Without start position, 'forward' starts at the beginning of the logfile and 'backward' at its end."
        ),
    ] = "forward",
    limit: Annotated[
        int,
        Query(
            description="Maximum number of lines to return (0 to 5000)."
        ),
    ] = DEFAULT_LIMIT,
    max_bytes: Annotated[
        int,
        Query(
            description="Soft cap on the summed size of the returned lines. At most 4 MiB are used."
        ),
    ] = DEFAULT_MAX_BYTES,
    levels: Annotated[
        str | None,
        Query(
            description="Comma separated list of the levels to return (DEBUG, INFO, WARNING, ERROR). Lines without a level are always returned. An empty value returns only those."
        ),
    ] = None,
    q: Annotated[
        str | None,
        Query(
            description="Text to search for. Only matching lines are returned."
        ),
    ] = None,
    regex: Annotated[
        bool,
        Query(
            description="The search text is a regular expression (Python syntax)."
        ),
    ] = False,
    case: Annotated[
        bool,
        Query(
            description="Match the case of the search text."
        ),
    ] = False,
    word: Annotated[
        bool, Query(description="Match whole words only.")
    ] = False,
    count_only: Annotated[
        bool,
        Query(
            description="Return no lines but count them. The limit and max_bytes are ignored."
        ),
    ] = False,
) -> LogLinesRequest

Read the query parameters of a request for log lines (a dependency of the /lines endpoints).

The payload logs and the pod logs are read with the same parameters (see build_lines_request).

Raises:

Type	Description
HTTPException	A parameter is invalid (400).

Returns:

Name	Type	Description
LogLinesRequest	LogLinesRequest	The request.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/functions.py

def lines_request_from_query(
    offset: Annotated[
        int | None,
        Query(description="Byte offset to start at. It is snapped back to the start of the line containing it."),
    ] = None,
    line: Annotated[
        int | None,
        Query(description="Line number (1-based) to start at. Alternative to the offset."),
    ] = None,
    direction: Annotated[
        str,
        Query(
            description="'forward' returns the lines from the start position on, "
            "'backward' the lines before it. Without start position, 'forward' starts "
            "at the beginning of the logfile and 'backward' at its end.",
        ),
    ] = "forward",
    limit: Annotated[int, Query(description="Maximum number of lines to return (0 to 5000).")] = DEFAULT_LIMIT,
    max_bytes: Annotated[
        int,
        Query(description="Soft cap on the summed size of the returned lines. At most 4 MiB are used."),
    ] = DEFAULT_MAX_BYTES,
    levels: Annotated[
        str | None,
        Query(
            description="Comma separated list of the levels to return (DEBUG, INFO, WARNING, ERROR). "
            "Lines without a level are always returned. An empty value returns only those.",
        ),
    ] = None,
    q: Annotated[str | None, Query(description="Text to search for. Only matching lines are returned.")] = None,
    regex: Annotated[bool, Query(description="The search text is a regular expression (Python syntax).")] = False,
    case: Annotated[bool, Query(description="Match the case of the search text.")] = False,
    word: Annotated[bool, Query(description="Match whole words only.")] = False,
    count_only: Annotated[
        bool,
        Query(description="Return no lines but count them. The limit and max_bytes are ignored."),
    ] = False,
) -> LogLinesRequest:
    """Read the query parameters of a request for log lines (a dependency of the `/lines` endpoints).

    The payload logs and the pod logs are read with the same parameters (see `build_lines_request`).

    Raises:
        HTTPException:
            A parameter is invalid (400).

    Returns:
        LogLinesRequest:
            The request.

    """

    try:
        return build_lines_request(
            offset=offset,
            line=line,
            direction=direction,
            limit=limit,
            max_bytes=max_bytes,
            levels=levels,
            q=q,
            regex=regex,
            case=case,
            word=word,
            count_only=count_only,
        )
    except ValueError as exc:
        raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(exc)) from exc

log_reader

Windowed reading, level filtering and searching of payload log files of any size.

A payload log can grow to several gigabytes. Nothing in this module reads a whole log into memory: the file is read in chunks, and a small in-memory index per log file makes every byte offset resolvable to a line number and a log level.

Terms used throughout this module:

• Offsets are byte offsets into the log file, never character positions.
• A line ends at a line feed. A trailing carriage return is not part of its text. A line is at most MAX_LINE_BYTES long (terminator included); longer lines are split at that length so that a file without line feeds cannot exhaust the memory.
• A record start is a line carrying a level marker (see RECORD_RE). The level of a line is the level of the nearest record start at or before it, CRITICAL counting as ERROR.
• Only complete lines are served. The unterminated end of the file is a line only once the payload is no longer running.

Classes:

Name	Description
LogChangedError	The log file was replaced while it was read.
LogQuery	A compiled search query.
LogLinesRequest	The parameters of a request for log lines.
IndexView	A consistent snapshot of a line index.
LogIndex	Incremental line index of one log file.

Functions:

Name	Description
get_index	Get the line index of a log file, creating it if necessary.
drop_index	Discard the line index of a log file, e.g. because the file is gone.
decode_line	Convert the bytes of a line to its text.
parse_levels	Convert the level filter of a request.
compile_query	Compile the search query of a request.
build_lines_request	Check and convert the parameters of a request for log lines.
read_log_info	Describe a log file and the progress of its index.
read_log_lines	Read a window of lines from a log file.

LogChangedError

Bases: Exception

The log file was replaced while it was read.

LogQuery dataclass

LogQuery(
    pattern: Pattern[str],
    needle: bytes | None = None,
    fold: bool = False,
    extras: tuple[bytes, ...] = (),
    exact: bool = False,
    word_left: bool = False,
    word_right: bool = False,
)

A compiled search query.

Attributes:

Name	Type	Description
pattern	Pattern[str]	Decides whether the text of a line matches.
needle	bytes | None	Bytes every matching line contains (lower case if fold is set). Used to skip lines quickly; None if no such bytes are known.
fold	bool	Whether the needle has to be searched ignoring the case of ASCII letters.
extras	tuple[bytes, ...]	Byte sequences that make the needle unreliable for a piece of the log.
exact	bool	Whether the needle is the whole query: where the needle is found, the query matches, provided the word boundaries are as required.
word_left	bool	Whether an exact needle must not directly follow a word character.
word_right	bool	Whether an exact needle must not be directly followed by a word character.

Methods:

Name	Description
matches	Check whether a line matches the query.

matches

matches(raw: bytes) -> bool

Check whether a line matches the query.

Parameters:

Name	Type	Description	Default
raw	bytes	The bytes of the line without the line feed.	required

Returns:

Name	Type	Description
bool	bool	True if the query matches somewhere in the line.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def matches(self, raw: bytes) -> bool:
    """Check whether a line matches the query.

    Args:
        raw (bytes):
            The bytes of the line without the line feed.

    Returns:
        bool:
            True if the query matches somewhere in the line.

    """

    return self.pattern.search(decode_line(raw)) is not None

LogLinesRequest dataclass

LogLinesRequest(
    offset: int | None = None,
    line: int | None = None,
    backward: bool = False,
    limit: int = DEFAULT_LIMIT,
    max_bytes: int = DEFAULT_MAX_BYTES,
    levels: int | None = None,
    query: LogQuery | None = None,
    count_only: bool = False,
)

The parameters of a request for log lines.

Attributes:

Name	Type	Description
offset	int | None	Byte offset to start at. It is snapped back to the start of its line.
line	int | None	1-based line number to start at (alternative to offset).
backward	bool	Return the lines before the start position instead of the lines from it on.
limit	int	Maximum number of lines to return.
max_bytes	int	Soft cap for the summed size of the returned lines.
levels	int | None	Bit mask of the level codes that pass. None lets every level pass.
query	LogQuery | None	The search query. None lets every line pass.
count_only	bool	Count the passing lines instead of returning them.

IndexView dataclass

IndexView(
    ident: tuple[int, int],
    size: int,
    complete: bool,
    indexed_bytes: int,
    lines: int,
    level: int,
    counts: tuple[int, ...],
    offsets: array,
    line_nos: array,
    levels: bytearray,
    forced: tuple[int, ...],
    checkpoints: int,
)

A consistent snapshot of a line index.

The checkpoint containers are shared with the index. They only ever grow, and a view must not look beyond its own checkpoints count.

Attributes:

Name	Type	Description
ident	tuple[int, int]	Device and inode of the indexed file.
size	int	Size of the file in bytes as far as known.
complete	bool	Whether the index has seen the file up to size.
indexed_bytes	int	End of the indexed lines.
lines	int	Number of indexed lines.
level	int	Level code in effect at indexed_bytes.
counts	tuple[int, ...]	Number of record starts per level code.
offsets	array	Byte offset of each checkpoint. Checkpoints are line starts.
line_nos	array	Number of lines before each checkpoint.
levels	bytearray	Level code in effect at each checkpoint.
forced	tuple[int, ...]	Numbers of the checkpoints that start an overlong line, which was split and therefore does not end with a line feed.
checkpoints	int	Number of checkpoints this view may use.

LogIndex

LogIndex(path: str)

Incremental line index of one log file.

The index knows the number of lines, the number of record starts per level and a sparse list of checkpoints (byte offset, line number, level in effect). It is extended by a background thread and held in memory only; it needs about 17 bytes per checkpoint, which is some 270 KiB per gigabyte of log.

Initialize the index.

Parameters:

Name	Type	Description	Default
path	str	The path of the log file.	required

Methods:

Name	Description
refresh	Bring the index up to date with the file and return a snapshot.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def __init__(self, path: str) -> None:
    """Initialize the index.

    Args:
        path (str):
            The path of the log file.

    """

    self.path = path
    self._lock = threading.Lock()
    self._changed = threading.Condition(self._lock)
    self._ident: tuple[int, int] | None = None
    self._generation = 0
    self._building = False
    self._clear()

refresh

refresh(
    stat_result: stat_result,
    wait: float = INDEX_WAIT_SECONDS,
) -> IndexView

Bring the index up to date with the file and return a snapshot.

Starts the background thread if the file has content the index has not seen, and waits for it for a limited time. The thread keeps running after this method returned.

Parameters:

Name	Type	Description	Default
stat_result	stat_result	The current status of the log file.	required
wait	float	The maximum time in seconds to wait for the index to catch up.	INDEX_WAIT_SECONDS

Returns:

Name	Type	Description
IndexView	IndexView	The snapshot. Its complete flag tells whether the index caught up.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def refresh(self, stat_result: os.stat_result, wait: float = INDEX_WAIT_SECONDS) -> IndexView:
    """Bring the index up to date with the file and return a snapshot.

    Starts the background thread if the file has content the index has not seen, and
    waits for it for a limited time. The thread keeps running after this method returned.

    Args:
        stat_result (os.stat_result):
            The current status of the log file.
        wait (float, optional):
            The maximum time in seconds to wait for the index to catch up.

    Returns:
        IndexView:
            The snapshot. Its `complete` flag tells whether the index caught up.

    """

    ident = (stat_result.st_dev, stat_result.st_ino)
    size = stat_result.st_size
    deadline = time.monotonic() + wait

    with self._changed:
        # The file was replaced or it shrank: start over.
        if ident != self._ident or size < self._indexed:
            self._ident = ident
            self._clear()

        started_at = -1
        while self._scanned < size and self._ident == ident:
            if not self._building:
                # A thread that ended without getting anywhere is not started again
                # by the same request.
                if started_at == self._scanned:
                    break
                started_at = self._scanned
                self._building = True
                threading.Thread(
                    target=self._build,
                    args=(self._generation,),
                    name="LogIndex",
                    daemon=True,
                ).start()
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            self._changed.wait(remaining)

        if self._ident != ident:
            # Another request found a newer file under the same path meanwhile:
            # nothing is known about the file of this request.
            return IndexView(
                ident=ident,
                size=size,
                complete=False,
                indexed_bytes=0,
                lines=0,
                level=0,
                counts=(0,) * len(LEVEL_NAMES),
                offsets=array("q", [0]),
                line_nos=array("q", [0]),
                levels=bytearray(1),
                forced=(),
                checkpoints=1,
            )

        return IndexView(
            ident=ident,
            size=max(size, self._scanned),
            complete=self._scanned >= size,
            indexed_bytes=self._indexed,
            lines=self._lines,
            level=self._level,
            counts=tuple(self._counts),
            offsets=self._offsets,
            line_nos=self._line_nos,
            levels=self._levels,
            forced=tuple(self._forced),
            checkpoints=len(self._offsets),
        )

get_index

get_index(path: str) -> LogIndex

Get the line index of a log file, creating it if necessary.

Parameters:

Name	Type	Description	Default
path	str	The path of the log file.	required

Returns:

Name	Type	Description
LogIndex	LogIndex	The index of the file.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def get_index(path: str) -> LogIndex:
    """Get the line index of a log file, creating it if necessary.

    Args:
        path (str):
            The path of the log file.

    Returns:
        LogIndex:
            The index of the file.

    """

    with _INDEXES_LOCK:
        index = _INDEXES.get(path)
        if index is None:
            index = _INDEXES[path] = LogIndex(path)
        return index

drop_index

drop_index(path: str) -> None

Discard the line index of a log file, e.g. because the file is gone.

Parameters:

Name	Type	Description	Default
path	str	The path of the log file.	required

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def drop_index(path: str) -> None:
    """Discard the line index of a log file, e.g. because the file is gone.

    Args:
        path (str):
            The path of the log file.

    """

    with _INDEXES_LOCK:
        _INDEXES.pop(path, None)

decode_line

decode_line(raw: bytes) -> str

Convert the bytes of a line to its text.

Parameters:

Name	Type	Description	Default
raw	bytes	The bytes of the line without the line feed.	required

Returns:

Name	Type	Description
str	str	The text of the line: without a trailing carriage return and with invalid UTF-8 replaced.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def decode_line(raw: bytes) -> str:
    """Convert the bytes of a line to its text.

    Args:
        raw (bytes):
            The bytes of the line without the line feed.

    Returns:
        str:
            The text of the line: without a trailing carriage return and with invalid
            UTF-8 replaced.

    """

    if raw.endswith(b"\r"):
        raw = raw[:-1]
    return raw.decode("utf-8", "replace")

parse_levels

parse_levels(value: str | None) -> int | None

Convert the level filter of a request.

Parameters:

Name	Type	Description	Default
value	str | None	Comma separated level names. None means all levels, an empty value means no level (only lines without a level pass).	required

Raises:

Type	Description
ValueError	If the value names an unknown level.

Returns:

Type	Description
int | None	int | None: The bit mask of the passing level codes, or None if every level passes.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def parse_levels(value: str | None) -> int | None:
    """Convert the level filter of a request.

    Args:
        value (str | None):
            Comma separated level names. None means all levels, an empty value means no
            level (only lines without a level pass).

    Raises:
        ValueError:
            If the value names an unknown level.

    Returns:
        int | None:
            The bit mask of the passing level codes, or None if every level passes.

    """

    if value is None:
        return None

    mask = 0
    for name in value.split(","):
        name = name.strip().upper()
        if not name:
            continue
        if name not in FILTER_LEVELS:
            message = "Unknown level -> '{}'. Allowed levels are {}.".format(name, ", ".join(FILTER_LEVELS))
            raise ValueError(message)
        mask |= 1 << FILTER_LEVELS[name]

    return None if mask == ALL_LEVELS_MASK else mask

compile_query

compile_query(
    q: str | None,
    *,
    regex: bool = False,
    case: bool = False,
    word: bool = False,
) -> LogQuery | None

Compile the search query of a request.

Parameters:

Name	Type	Description	Default
q	str | None	The search text. None or an empty text means no query.	required
regex	bool	Treat the text as a regular expression (Python syntax) instead of literally.	False
case	bool	Match the case exactly.	False
word	bool	Match whole words only: neither end of the match may lie between two word characters.	False

Raises:

Type	Description
ValueError	If the text is not a valid regular expression.

Returns:

Type	Description
LogQuery | None	LogQuery | None: The compiled query, or None if there is nothing to search for.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def compile_query(q: str | None, *, regex: bool = False, case: bool = False, word: bool = False) -> LogQuery | None:
    """Compile the search query of a request.

    Args:
        q (str | None):
            The search text. None or an empty text means no query.
        regex (bool, optional):
            Treat the text as a regular expression (Python syntax) instead of literally.
        case (bool, optional):
            Match the case exactly.
        word (bool, optional):
            Match whole words only: neither end of the match may lie between two word
            characters.

    Raises:
        ValueError:
            If the text is not a valid regular expression.

    Returns:
        LogQuery | None:
            The compiled query, or None if there is nothing to search for.

    """

    if not q:
        return None

    expression = q if regex else re.escape(q)
    if word:
        boundary = r"(?:(?<!\w)|(?!\w))"
        expression = boundary + "(?:" + expression + ")" + boundary
    try:
        pattern = re.compile(expression, 0 if case else re.IGNORECASE)
    except (re.error, RecursionError, OverflowError) as exc:
        message = "Invalid regular expression -> {}".format(exc)
        raise ValueError(message) from exc

    if regex or "\n" in q or "\ufffd" in q:
        return LogQuery(pattern=pattern)

    # Without regular expression every matching line contains the text itself. With the case
    # ignored this is only safe for the ASCII part of the text, where the case rules are simple.
    if q.isascii() and "\r" not in q:
        # Pure ASCII text is found reliably in the bytes of the log. Only the word boundaries
        # are left to check, at the ends of the text that are word characters themselves.
        return LogQuery(
            pattern=pattern,
            needle=(q if case else q.lower()).encode("ascii"),
            fold=not case,
            extras=() if case else _fold_extras(q.lower()),
            exact=True,
            word_left=word and _is_word(q[0]),
            word_right=word and _is_word(q[-1]),
        )
    if case:
        try:
            return LogQuery(pattern=pattern, needle=q.encode("utf-8"))
        except UnicodeEncodeError:
            return LogQuery(pattern=pattern)

    run = max(re.findall(r"[\x00-\x7f]+", q), key=len, default="").lower()
    if not run:
        return LogQuery(pattern=pattern)

    return LogQuery(pattern=pattern, needle=run.encode("ascii"), fold=True, extras=_fold_extras(run))

build_lines_request

build_lines_request(
    *,
    offset: int | None = None,
    line: int | None = None,
    direction: str = "forward",
    limit: int = DEFAULT_LIMIT,
    max_bytes: int = DEFAULT_MAX_BYTES,
    levels: str | None = None,
    q: str | None = None,
    regex: bool = False,
    case: bool = False,
    word: bool = False,
    count_only: bool = False,
) -> LogLinesRequest

Check and convert the parameters of a request for log lines.

Parameters:

Name	Type	Description	Default
offset	int | None	The byte offset to start at.	None
line	int | None	The 1-based line number to start at. Only one of offset and line may be given.	None
direction	str	"forward" or "backward".	'forward'
limit	int	The maximum number of lines to return (0 to MAX_LIMIT).	DEFAULT_LIMIT
max_bytes	int	The soft cap for the summed size of the returned lines.	DEFAULT_MAX_BYTES
levels	str | None	The level filter (see parse_levels).	None
q	str | None	The search text (see compile_query).	None
regex	bool	Treat the search text as a regular expression.	False
case	bool	Match the case of the search text.	False
word	bool	Match whole words only.	False
count_only	bool	Count the passing lines instead of returning them.	False

Raises:

Type	Description
ValueError	If a parameter is invalid. The message says which one and why.

Returns:

Name	Type	Description
LogLinesRequest	LogLinesRequest	The request.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def build_lines_request(
    *,
    offset: int | None = None,
    line: int | None = None,
    direction: str = "forward",
    limit: int = DEFAULT_LIMIT,
    max_bytes: int = DEFAULT_MAX_BYTES,
    levels: str | None = None,
    q: str | None = None,
    regex: bool = False,
    case: bool = False,
    word: bool = False,
    count_only: bool = False,
) -> LogLinesRequest:
    """Check and convert the parameters of a request for log lines.

    Args:
        offset (int | None, optional):
            The byte offset to start at.
        line (int | None, optional):
            The 1-based line number to start at. Only one of `offset` and `line` may be given.
        direction (str, optional):
            "forward" or "backward".
        limit (int, optional):
            The maximum number of lines to return (0 to `MAX_LIMIT`).
        max_bytes (int, optional):
            The soft cap for the summed size of the returned lines.
        levels (str | None, optional):
            The level filter (see `parse_levels`).
        q (str | None, optional):
            The search text (see `compile_query`).
        regex (bool, optional):
            Treat the search text as a regular expression.
        case (bool, optional):
            Match the case of the search text.
        word (bool, optional):
            Match whole words only.
        count_only (bool, optional):
            Count the passing lines instead of returning them.

    Raises:
        ValueError:
            If a parameter is invalid. The message says which one and why.

    Returns:
        LogLinesRequest:
            The request.

    """

    message = None
    if offset is not None and line is not None:
        message = "Only one of 'offset' and 'line' may be given."
    elif offset is not None and offset < 0:
        message = "The offset must not be negative."
    elif line is not None and line < 1:
        message = "The line must be 1 or higher."
    elif direction not in ("forward", "backward"):
        message = "Unknown direction -> '{}'. Allowed directions are forward, backward.".format(direction)
    elif not 0 <= limit <= MAX_LIMIT:
        message = "The limit must be between 0 and {}.".format(MAX_LIMIT)
    elif max_bytes < 0:
        message = "The value of max_bytes must not be negative."
    if message:
        raise ValueError(message)

    return LogLinesRequest(
        offset=offset,
        line=line,
        backward=direction == "backward",
        limit=limit,
        max_bytes=max_bytes,
        levels=parse_levels(levels),
        query=compile_query(q, regex=regex, case=case, word=word),
        count_only=count_only,
    )

read_log_info

read_log_info(
    path: str,
    running: bool,
    wait: float = INDEX_WAIT_SECONDS,
) -> dict

Describe a log file and the progress of its index.

Parameters:

Name	Type	Description	Default
path	str	The path of the log file.	required
running	bool	Whether the payload writing the log is still running.	required
wait	float	The maximum time in seconds to wait for the index to catch up.	INDEX_WAIT_SECONDS

Raises:

Type	Description
FileNotFoundError	If the log file does not exist.

Returns:

Name	Type	Description
dict	dict	The fields of the info response.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def read_log_info(path: str, running: bool, wait: float = INDEX_WAIT_SECONDS) -> dict:
    """Describe a log file and the progress of its index.

    Args:
        path (str):
            The path of the log file.
        running (bool):
            Whether the payload writing the log is still running.
        wait (float, optional):
            The maximum time in seconds to wait for the index to catch up.

    Raises:
        FileNotFoundError:
            If the log file does not exist.

    Returns:
        dict:
            The fields of the info response.

    """

    file, view = _open_log(path, wait)
    with file:
        lines = view.lines
        counts = list(view.counts)
        indexed = view.indexed_bytes
        if _has_tail(view, running):
            file.seek(indexed)
            found = RECORD_RE.findall(file.read(view.size - indexed))
            if found:
                counts[LEVEL_CODES[found[0]]] += 1
            lines += 1
            indexed = view.size

    return {
        "version": CONTRACT_VERSION,
        "size": view.size,
        "indexed_bytes": indexed,
        "indexing": not view.complete,
        "lines": lines,
        "counts": {name: counts[code] for name, code in sorted(FILTER_LEVELS.items(), key=lambda item: -item[1])},
        "running": running,
        "record_start": RECORD_START_PATTERN,
    }

read_log_lines

read_log_lines(
    path: str,
    running: bool,
    request: LogLinesRequest,
    cancel: Event | None = None,
    *,
    wait: float | None = None,
    budget_seconds: float | None = None,
    budget_bytes: int | None = None,
) -> dict

Read a window of lines from a log file.

Parameters:

Name	Type	Description	Default
path	str	The path of the log file.	required
running	bool	Whether the payload writing the log is still running.	required
request	LogLinesRequest	What to read.	required
cancel	Event | None	Set by the caller when the result is no longer wanted; the scan then stops.	None
wait	float | None	The maximum time in seconds to wait for the index to catch up.	None
budget_seconds	float | None	The time after which the scan stops and reports how far it got.	None
budget_bytes	int | None	The number of scanned bytes after which the scan stops.	None

Raises:

Type	Description
FileNotFoundError	If the log file does not exist.

Returns:

Name	Type	Description
dict	dict	The fields of the lines response.

Source code in packages/pyxecm/src/pyxecm_api/v1_payload/log_reader.py

def read_log_lines(
    path: str,
    running: bool,
    request: LogLinesRequest,
    cancel: threading.Event | None = None,
    *,
    wait: float | None = None,
    budget_seconds: float | None = None,
    budget_bytes: int | None = None,
) -> dict:
    """Read a window of lines from a log file.

    Args:
        path (str):
            The path of the log file.
        running (bool):
            Whether the payload writing the log is still running.
        request (LogLinesRequest):
            What to read.
        cancel (threading.Event | None, optional):
            Set by the caller when the result is no longer wanted; the scan then stops.
        wait (float | None, optional):
            The maximum time in seconds to wait for the index to catch up.
        budget_seconds (float | None, optional):
            The time after which the scan stops and reports how far it got.
        budget_bytes (int | None, optional):
            The number of scanned bytes after which the scan stops.

    Raises:
        FileNotFoundError:
            If the log file does not exist.

    Returns:
        dict:
            The fields of the lines response.

    """

    budget_seconds = SCAN_BUDGET_SECONDS if budget_seconds is None else budget_seconds
    deadline = time.monotonic() + budget_seconds
    wait = min(INDEX_WAIT_SECONDS if wait is None else wait, budget_seconds)

    # A file that is replaced while it is read is read again, once.
    for attempt in range(2):
        file, view = _open_log(path, wait)
        with file:
            tail = _has_tail(view, running)
            scan = _Scan(
                file,
                view,
                request,
                tail=tail,
                deadline=deadline,
                budget_bytes=SCAN_BUDGET_BYTES if budget_bytes is None else budget_bytes,
                cancel=cancel,
            )
            try:
                position = scan.resolve()
                if position is None or (
                    # Nothing of what a forward request asks for is indexed yet.
                    not view.complete and not request.backward and position >= scan.end
                ):
                    covered = False
                    if position is None:
                        position = view.indexed_bytes if request.offset is None else min(request.offset, view.size)
                        if request.line is None and request.offset is None:
                            position = view.size
                    start = end = position
                else:
                    covered = True
                    start, end = scan.run(position)
            except LogChangedError:
                if attempt:
                    raise
                drop_index(path)
                continue
        break

    return {
        "lines": scan.lines if covered else [],
        "matched": scan.matched if covered else 0,
        "start": start,
        "end": end,
        "bof": start == 0,
        "eof": view.complete and end >= scan.end,
        "size": view.size,
        "indexing": not view.complete,
        "indexed_bytes": view.size if tail else view.indexed_bytes,
        "running": running,
    }

Terminal

router

Define a websocket to connect to a shell session in a pod.

Functions:

Name	Description
local_executable	Find a command to run in the Customizer container.
valid_cwd	Check the folder a terminal is asked to start in.
ws_terminal	Websocket to connect to a shell session in a pod.

local_executable

local_executable(command: str) -> str

Find a command to run in the Customizer container.

A path that does not exist (anymore) falls back to the command's name on the PATH: the manager asked for k9s at /root/.local/bin/k9s, where older images installed it.

Parameters:

Name	Type	Description	Default
command	str	command name, or path of the executable.	required

Returns:

Name	Type	Description
str	str	The executable to run.

Source code in packages/pyxecm/src/pyxecm_api/terminal/router.py

def local_executable(command: str) -> str:
    """Find a command to run in the Customizer container.

    A path that does not exist (anymore) falls back to the command's name on the PATH: the manager
    asked for k9s at /root/.local/bin/k9s, where older images installed it.

    Args:
        command (str): command name, or path of the executable.

    Returns:
        str: The executable to run.

    """

    if os.sep in command and not os.access(command, os.X_OK):
        return shutil.which(os.path.basename(command)) or command
    return command

valid_cwd

valid_cwd(cwd: str | None) -> bool

Check the folder a terminal is asked to start in.

Source code in packages/pyxecm/src/pyxecm_api/terminal/router.py

def valid_cwd(cwd: str | None) -> bool:
    """Check the folder a terminal is asked to start in."""

    return cwd is None or (cwd.startswith("/") and "\0" not in cwd and len(cwd) <= MAX_PATH_LENGTH)

ws_terminal async

ws_terminal(
    websocket: WebSocket,
    pod: str = Query(...),
    command: str = Query(...),
    container: str = Query(None),
    cwd: str = Query(None),
) -> None

Websocket to connect to a shell session in a pod.

Parameters:

Name	Type	Description	Default
websocket	WebSocket	WebSocket to connect to the shell session.	required
pod	str	pod name to connect to.	Query(...)
command	str	command to be executed.	Query(...)
container	str	container name to connect to.	Query(None)
cwd	str	folder to start the command in (absolute).	Query(None)

Source code in packages/pyxecm/src/pyxecm_api/terminal/router.py

@router.websocket("/ws/terminal")
async def ws_terminal(
    websocket: WebSocket,
    pod: str = Query(...),
    command: str = Query(...),
    container: str = Query(None),
    cwd: str = Query(None),
) -> None:
    """Websocket to connect to a shell session in a pod.

    Args:
        websocket (WebSocket): WebSocket to connect to the shell session.
        pod (str): pod name to connect to.
        command (str): command to be executed.
        container (str, optional): container name to connect to.
        cwd (str, optional): folder to start the command in (absolute).

    """
    await websocket.accept()

    # A pod or container name is checked, so that it cannot be taken for an option of kubectl.
    if (
        (pod != "customizer" and not is_pod_name(pod))
        or (container is not None and not is_container_name(container))
        or not valid_cwd(cwd)
    ):
        await websocket.send_text("Invalid pod, container or folder")
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return

    try:
        # Wait for the first message to be the token
        token = await websocket.receive_text()

        # get_current_user is sync (blocking HTTP call to OTDS) - run it off the event loop
        user = await asyncio.to_thread(get_current_user, token=token, api_key=token)
        await get_authorized_user(user)

    except HTTPException as e:
        await websocket.send_text("Invalid Token" if e.status_code == status.HTTP_401_UNAUTHORIZED else e.detail)
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return

    except WebSocketDisconnect:
        return

    container = ["-c", container] if container else []

    if pod == "customizer":
        process = [local_executable(command)]
    elif cwd:
        process = ["kubectl", "exec", "-it", pod, *container, "--", "sh", "-c", CWD_SCRIPT, "sh", cwd, command]
    else:
        process = ["kubectl", "exec", "-it", pod, *container, "--", command]

    pid, fd = pty.fork()
    import fcntl
    import struct
    import termios

    if pid == 0:
        # The child is a copy of the API process: it becomes the command, with the pty as its
        # terminal, and must never return into the code of the API.
        if pod == "customizer" and cwd:
            with contextlib.suppress(OSError):
                os.chdir(cwd)
        try:
            os.execvp(process[0], process)  # noqa: S606
        except OSError as error:
            os.write(2, "Cannot start '{}': {}\r\n".format(process[0], error).encode())
        os._exit(127)

    async def read_from_pty() -> None:
        loop = asyncio.get_event_loop()
        try:
            while True:
                data = await loop.run_in_executor(None, os.read, fd, 1024)
                if not data:
                    break  # The command ended (Linux raises an OSError instead).
                await websocket.send_text(data.decode(errors="ignore"))
        except Exception:  # noqa: S110
            pass  # PTY closed or WebSocket failed

    async def write_to_pty() -> None:
        try:
            while True:
                data = await websocket.receive_text()
                if data.startswith("{") and data.endswith("}"):
                    message = json.loads(data)
                    if message.get("type") == "resize":
                        rows = message.get("rows")
                        cols = message.get("cols")

                        packed_size = struct.pack("HHHH", rows, cols, 0, 0)
                        fcntl.ioctl(fd, termios.TIOCSWINSZ, packed_size)

                else:
                    os.write(fd, data.encode())
        except Exception:  # noqa: S110
            pass

    try:
        # Launch read/write tasks
        read_task = asyncio.create_task(read_from_pty())
        write_task = asyncio.create_task(write_to_pty())

        _done, pending = await asyncio.wait([read_task, write_task], return_when=asyncio.FIRST_COMPLETED)

        # Cancel other task
        for task in pending:
            task.cancel()
    finally:
        with contextlib.suppress(Exception):
            await websocket.close(code=status.WS_1008_POLICY_VIOLATION)

        with contextlib.suppress(Exception):
            os.kill(pid, signal.SIGKILL)

        # Reap the child, so that no zombie process is left behind.
        with contextlib.suppress(Exception):
            await asyncio.to_thread(os.waitpid, pid, 0)

        with contextlib.suppress(Exception):
            os.close(fd)