rs_dpr_service/main.md

<< Back to index

rs dpr service main module.

DatabaseJobFormatError

Bases: Exception

Exception raised when an error occurred during the init of a provider.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

class DatabaseJobFormatError(Exception):
    """Exception raised when an error occurred during the init of a provider."""

JobsFormatError

Bases: Exception

Exception raised when an error occurred during the init of a provider.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

class JobsFormatError(Exception):
    """Exception raised when an error occurred during the init of a provider."""

app_lifespan(fastapi_app) async

Lifespann app to be implemented with start up / stop logic

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

@asynccontextmanager
async def app_lifespan(fastapi_app: FastAPI):
    """Lifespann app to be implemented with start up / stop logic"""
    logger.info("Starting up the application...")
    # Create jobs table
    process_manager = init_db()

    fastapi_app.state.process_manager = process_manager
    # fastapi_app.state.db_table = db.table("jobs")
    # fastapi_app.state.dask_cluster = cluster

    # Yield control back to the application (this is where the app will run)
    yield

    # Shutdown logic (cleanup)
    logger.info("Shutting down the application...")
    logger.info("Application gracefully stopped...")

build_cluster_info(data)

This function handles missing parameters from request, hence properly creates a ClusterInfo object.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

def build_cluster_info(data: dict) -> ClusterInfo:
    """This function handles missing parameters from request, hence properly creates a ClusterInfo object."""

    # Mandatory fields
    try:
        jupyter_token = data["jupyter_token"]
        dask_gateway_address = data["dask_gateway_address"]
        cluster_label = data["cluster_label"]
    except KeyError as error:
        raise HTTPException(status_code=400, detail=f"Missing required field: {str(error)}") from error

    # Optional fields
    cluster_instance = data.get("cluster_instance", "")

    return ClusterInfo(
        jupyter_token=jupyter_token,
        dask_gateway_address=dask_gateway_address,
        cluster_label=cluster_label,
        cluster_instance=cluster_instance,
    )

delete_job_endpoint(request, job_id=Annotated[str, Path(..., title='The ID of the job to delete')]) async

Deletes a specific job from the database.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

@router.delete("/dpr/jobs/{job_id}")
async def delete_job_endpoint(request: Request, job_id=Annotated[str, Path(..., title="The ID of the job to delete")]):
    """Deletes a specific job from the database."""

    # Send a dask distributed event for the cancellation of the job
    cancel_event = distributed.Event(CANCEL_JOB.format(job_id=job_id))
    if cancel_event.client:
        cancel_event.set()

    try:
        job = app.state.process_manager.get_job(job_id)
    # Handle case when job_id is not found
    except JobNotFoundError:  # pylint: disable=W0718
        return JSONResponse(status_code=HTTP_404_NOT_FOUND, content=f"Job with ID {job_id} not found")

    app.state.process_manager.delete_job(job_id)

    # Create job response with a status message to confirm the job deletion
    job["message"] = f"Job {job_id} deleted successfully"
    formatted_job_data = format_job_data(job)
    validate_response(request, formatted_job_data)
    return JSONResponse(status_code=HTTP_200_OK, content=formatted_job_data)

execute_process(request, resource) async

Used to execute processing jobs.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

@router.post("/dpr/processes/{resource}/execution")
async def execute_process(request: Request, resource: str):  # pylint: disable=unused-argument
    """Used to execute processing jobs."""

    # Validate request payload. This will create the 'http receive' fastapi span
    valid_body = await validate_request(request)

    # Run business logic in our own span
    with start_span(__name__, f"execute_processor_{resource}") as span:
        try:
            # Check that the input resource exists
            if resource not in api.config["resources"]:
                err_msg = f"Process {resource!r} not found"
                span.set_status(StatusCode.ERROR, err_msg)
                return JSONResponse(status_code=HTTP_404_NOT_FOUND, content=err_msg)

            cluster_info = build_cluster_info(valid_body)

            processor_name = api.config["resources"][resource]["processor"]["name"]
            if processor_name in processor_types:
                processor_type = processor_types[processor_name]
                # Asynchronously execute the dpr process in the dask cluster
                _, dpr_status = await processor_type(app.state.process_manager, cluster_info).execute(
                    valid_body,
                )  # type: ignore

                # Get identifier of the current job
                status_dict = {
                    "accepted": HTTP_201_CREATED,
                    "running": HTTP_201_CREATED,
                    "successful": HTTP_201_CREATED,
                    "failed": HTTP_500_INTERNAL_SERVER_ERROR,
                    "dismissed": HTTP_500_INTERNAL_SERVER_ERROR,
                }
                id_key = [status for status in status_dict if status in dpr_status][0]
                formatted_job_data = format_job_data(app.state.process_manager.get_job(dpr_status[id_key]))
                validate_response(request, formatted_job_data, HTTP_201_CREATED)
                span.set_status(StatusCode.OK, str(formatted_job_data))
                return JSONResponse(status_code=HTTP_201_CREATED, content=formatted_job_data)

            err_msg = f"Processor {processor_name!r} not found"
            span.set_status(StatusCode.ERROR, err_msg)
            return JSONResponse(status_code=HTTP_404_NOT_FOUND, content=err_msg)

        except Exception as e:
            record_error(span, e)
            raise

filter_enabled_resources(config)

Return the usual list of processors in local mode, and a specific one on cluster.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

def filter_enabled_resources(config: dict) -> dict:
    """Return the usual list of processors in local mode, and a specific one on cluster."""

    enabled = get_enabled_processors()

    # For localmode use the default configuration
    if enabled is None:
        return config

    config["resources"] = {
        resource_name: resource_config
        for resource_name, resource_config in config.get("resources", {}).items()
        if resource_config.get("processor", {}).get("name") in enabled
    }
    return config

format_job_data(job)

Method to apply reformatting on job data to make it compliant with OGC (process) standards Args: job: information on a specific job to fromat: the job must have the same attributes than the columns from the PostgreSql database Result: reformatted and validated job_data variable to put in the response

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

def format_job_data(job: dict):
    """
    Method to apply reformatting on job data to make it compliant with OGC (process) standards
    Args:
        job: information on a specific job to fromat: the job must have the same attributes
        than the columns from the PostgreSql database
    Result:
        reformatted and validated job_data variable to put in the response
    """
    # Check that the input job have the same struture as the jobs contained in the PostgreSQL database
    if "identifier" not in job:
        raise DatabaseJobFormatError(
            """Input job must have the same structure than the jobs stored in the """
            """PostgreSql database: attribute 'identifier' is missing""",
        )
    job_data = copy.deepcopy(job)
    # Rename attribute "identifier" to be compliant with OGC standards
    job_data[JOB_ATTRS_MAPPING["identifier"]] = job_data.pop("identifier")
    # Remove attributes which should not be part of the response
    for attr in OGC_UNCOMPLIANT_JOB_ATTRS:
        if attr in job_data:
            job_data.pop(attr)
    for key, value in job_data.items():
        # Reformat datetime object to string
        if isinstance(value, datetime):
            job_data[key] = value.strftime("%Y-%m-%dT%H:%M:%SZ")
    # Remove "finished" attribute if its value is None
    if "finished" in job_data and job_data.get("finished") is None:
        job_data.pop("finished")
    return job_data

format_jobs_data(jobs)

Method validate information on all existing jobs

Parameters:

Name	Type	Description	Default
jobs	dict	information on all existing jobs	required

Result: reformatted and validated jobs_data variable to provide to the response

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

def format_jobs_data(jobs: dict):
    """
    Method validate information on all existing jobs

    Args:
        jobs: information on all existing jobs
    Result:
        reformatted and validated jobs_data variable to provide to the response
    """
    if not isinstance(jobs, dict):
        raise JobsFormatError("Expected a dictionary as input")
    if "jobs" not in jobs:
        raise JobsFormatError("Invalid format for input jobs: missing 'jobs' key")
    jobs_data = copy.deepcopy(jobs)
    # Add "links" mandatory field to the response
    jobs_data.update(
        {
            "links": [
                {
                    "href": "string",
                    "rel": "service",
                    "type": "application/json",
                    "hreflang": "en",
                    "title": "List of jobs",
                },
            ],
        },
    )
    # Remove SQLAlchemy _sa_instance_state objects and convert datetime
    for i, job_data in enumerate(jobs_data["jobs"]):
        jobs_data["jobs"][i] = format_job_data(job_data)
    return jobs_data

get_config_contents()

Return the pygeoapi configuration yaml file contents.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

def get_config_contents() -> dict:
    """Return the pygeoapi configuration yaml file contents."""
    # Open the configuration file
    with open(get_config_path(), encoding="utf8") as opened:
        contents = opened.read()

        # Replace env vars by their value
        contents = Template(contents).substitute(os.environ)

        # Parse contents as yaml
        config = yaml.safe_load(contents)
        return filter_enabled_resources(config)

get_config_path()

Return the pygeoapi configuration path and set the PYGEOAPI_CONFIG env var accordingly.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

def get_config_path() -> pathlib.Path:
    """Return the pygeoapi configuration path and set the PYGEOAPI_CONFIG env var accordingly."""
    path = pathlib.Path(__file__).parent.parent / "config" / "geoapi.yaml"
    os.environ["PYGEOAPI_CONFIG"] = str(path)
    return path

get_enabled_processors()

Return the desired processors. These are set in DPR_ENABLED_PROCESSORS by deployment in rs-helm.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

def get_enabled_processors() -> set[str] | None:
    """Return the desired processors. These are set in DPR_ENABLED_PROCESSORS by deployment in rs-helm."""
    raw = os.environ.get("DPR_ENABLED_PROCESSORS")
    if raw is None:
        return None
    return {item.strip() for item in raw.split(",") if item.strip()}

get_job_logs_endpoint(request, job_id=Annotated[str, Path(..., title='The ID of the job')]) async

Used to stream logs of a processing job.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

@router.get("/dpr/jobs/{job_id}/logs")
async def get_job_logs_endpoint(request: Request, job_id=Annotated[str, Path(..., title="The ID of the job")]):
    """Used to stream logs of a processing job."""

    # Verify job exists
    try:
        app.state.process_manager.get_job(job_id)
    except JobNotFoundError:
        return JSONResponse(status_code=HTTP_404_NOT_FOUND, content=f"Job with ID {job_id} not found")

    queue: asyncio.Queue[str] = asyncio.Queue()
    job_log_handler.queues.setdefault(job_id, []).append(queue)

    async def log_generator():
        try:
            while True:
                if await request.is_disconnected():
                    break

                try:
                    # Wait for log message
                    msg = await asyncio.wait_for(queue.get(), timeout=2.0)
                    for line in msg.split("\n"):
                        yield f"data: {line}\n\n"
                except TimeoutError:
                    # Check if job is still running
                    try:
                        # Send a keepalive message to keep the connection alive (prevents proxy timeouts)
                        yield ": keepalive\n\n"
                        job_status = app.state.process_manager.get_job(job_id)
                        if job_status.get("status") in ["successful", "failed", "dismissed"]:
                            # job finished. drain remaining queue items (if any)
                            while not queue.empty():
                                yield f"data: {queue.get_nowait()}\n\n"
                            break
                    except Exception as err:  # pylint: disable=broad-exception-caught
                        # Do not break the stream if there's a transient DB error
                        logger.warning(f"Error checking job status during log stream: {err}")
                        continue
        finally:
            job_log_handler.queues[job_id].remove(queue)
            if not job_log_handler.queues[job_id]:
                del job_log_handler.queues[job_id]

    return StreamingResponse(log_generator(), media_type="text/event-stream")

get_job_status_endpoint(request, job_id=Annotated[str, Path(..., title='The ID of the job')]) async

Used to get status of processing job.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

@router.get("/dpr/jobs/{job_id}")
async def get_job_status_endpoint(request: Request, job_id=Annotated[str, Path(..., title="The ID of the job")]):
    """Used to get status of processing job."""
    try:
        job = app.state.process_manager.get_job(job_id)
    except JobNotFoundError:  # pylint: disable=W0718
        # Handle case when job_id is not found
        return JSONResponse(status_code=HTTP_404_NOT_FOUND, content=f"Job with ID {job_id} not found")

    formatted_job_data = format_job_data(job)
    validate_response(request, formatted_job_data)
    return JSONResponse(status_code=HTTP_200_OK, content=formatted_job_data)

get_jobs_endpoint(request) async

Returns the status of all jobs from database.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

@router.get("/dpr/jobs")
async def get_jobs_endpoint(request: Request):
    """Returns the status of all jobs from database."""
    try:
        # Generate an output conform to OGC process specifications
        formatted_jobs_data = format_jobs_data(app.state.process_manager.get_jobs())
        validate_response(request, formatted_jobs_data)
        return JSONResponse(status_code=HTTP_200_OK, content=formatted_jobs_data)
    except Exception as e:  # pylint: disable=W0718
        # Handle exceptions and return an appropriate error message
        return JSONResponse(status_code=HTTP_404_NOT_FOUND, content=str(e))

get_processes(request) async

Returns list of all available processes from config.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

@router.get("/dpr/processes")
async def get_processes(request: Request):
    """Returns list of all available processes from config."""
    processes = {
        "processes": [],
        "links": [
            {"href": str(request.url), "rel": "self", "type": "application/json", "title": "List of processes"},
        ],
    }
    for resource in api.config["resources"]:
        processes["processes"].append(
            {
                "id": api.config["resources"][resource]["processor"]["name"],
                "version": "1.0.0",
            },
        )
    validate_response(request, processes)
    return JSONResponse(status_code=HTTP_200_OK, content=processes)

get_resource(request, resource) async

Should return info about a specific resource.

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

@router.get("/dpr/processes/{resource}")
async def get_resource(request: Request, resource: str):
    """Should return info about a specific resource."""
    with start_span(__name__, "tasktable"):

        # Check that the input resource exists
        if resource not in api.config["resources"]:
            return JSONResponse(status_code=HTTP_404_NOT_FOUND, content=f"Process {resource!r} not found")

        cluster_info = build_cluster_info(dict(request.query_params))

        processor_name = api.config["resources"][resource]["processor"]["name"]
        if processor_name in processor_types:
            processor_type = processor_types[processor_name]
            task_table = await processor_type(app.state.process_manager, cluster_info).get_tasktable()
            return JSONResponse(status_code=HTTP_200_OK, content=task_table)

init_db(pause=3, timeout=None)

Initialize the PostgreSQL database connection and sets up required table and ENUM type.

This function constructs the database URL using environment variables for PostgreSQL credentials, host, port, and database name. It then creates an SQLAlchemy engine and registers the ENUM type JobStatus and the 'job' tables if they don't already exist.

Environment Variables

• POSTGRES_USER: Username for database authentication.
• POSTGRES_PASSWORD: Password for the database.
• POSTGRES_HOST: Hostname of the PostgreSQL server.
• POSTGRES_PORT: Port number of the PostgreSQL server.
• POSTGRES_DB: Database name.

Parameters:

Name	Type	Description	Default
pause	int	pause in seconds to wait for the database connection.	3
timeout	int | None	timeout in seconds to wait for the database connection.	None

Returns:

Type	Description
PostgreSQLManager	PostgreSQLManager instance

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

def init_db(pause: int = 3, timeout: int | None = None) -> PostgreSQLManager:
    """Initialize the PostgreSQL database connection and sets up required table and ENUM type.

    This function constructs the database URL using environment variables for PostgreSQL
    credentials, host, port, and database name. It then creates an SQLAlchemy engine and
    registers the ENUM type JobStatus and the 'job' tables if they don't already exist.

    Environment Variables:
        - POSTGRES_USER: Username for database authentication.
        - POSTGRES_PASSWORD: Password for the database.
        - POSTGRES_HOST: Hostname of the PostgreSQL server.
        - POSTGRES_PORT: Port number of the PostgreSQL server.
        - POSTGRES_DB: Database name.

    Args:
        pause: pause in seconds to wait for the database connection.
        timeout: timeout in seconds to wait for the database connection.

    Returns:
        PostgreSQLManager instance
    """
    manager_def = api.config["manager"]
    if not manager_def or not isinstance(manager_def, dict) or not isinstance(manager_def["connection"], dict):
        message = "Error reading the manager definition for pygeoapi PostgreSQL Manager"
        logger.error(message)
        raise RuntimeError(message)
    connection = manager_def["connection"]

    # Create SQL Alchemy engine
    engine = get_engine(driver_name="postgresql+psycopg2", **connection)

    while True:
        try:
            # This registers the ENUM type and creates the jobs table if they do not exist
            Base.metadata.create_all(bind=engine)
            logger.info(f"Reached {engine.url!r}")
            logger.info("Database table and ENUM type created successfully.")
            break

        # It fails if the database is unreachable. Wait a few seconds and try again.
        except SQLAlchemyError:
            logger.warning(f"Trying to reach {engine.url!r}")

            # Sleep for n seconds and raise exception if timeout is reached.
            if timeout is not None:
                timeout -= pause
                if timeout < 0:
                    raise
            sleep(pause)

    # Initialize PostgreSQLManager with the manager configuration
    return PostgreSQLManager(manager_def)

init_pygeoapi()

Init pygeoapi

Source code in docs/rs-dpr-service/rs_dpr_service/main.py

def init_pygeoapi() -> API:
    """Init pygeoapi"""
    return API(get_config_contents(), "")