rs_server_catalog/app.md

<< Back to index

RS-Server STAC catalog based on stac-fastapi-pgstac.

add_parameter_owner_id(parameters)

Add the owner id dictionnary to the parameter list.

Parameters:

Name	Type	Description	Default
parameters	list[dict]	the parameters list	required

Returns: dict: the new parameters list with the owner id parameter.

Source code in docs/rs-server/services/catalog/rs_server_catalog/app.py

def add_parameter_owner_id(parameters: list[dict]) -> list[dict]:
    """Add the owner id dictionnary to the parameter list.

    Args:
        parameters (list[dict]): the parameters list
        where we want to add the owner id parameter.
    Returns:
        dict: the new parameters list with the owner id parameter.
    """
    description = "Catalog owner id"
    to_add = {
        "description": description,
        "required": False,
        "schema": {"type": "string", "title": description, "description": description},
        "name": "owner_id",
        "in": "path",
    }
    parameters.append(to_add)
    return parameters

just_for_the_lock_icon(apikey_value='') async

Dummy function to add a lock icon in Swagger to enter an API key.

Source code in docs/rs-server/services/catalog/rs_server_catalog/app.py

async def just_for_the_lock_icon(
    apikey_value: Annotated[str, Security(APIKEY_AUTH_HEADER)] = "",  # pylint: disable=unused-argument
):
    """Dummy function to add a lock icon in Swagger to enter an API key."""

lifespan(my_app) async

The lifespan function.

Source code in docs/rs-server/services/catalog/rs_server_catalog/app.py

@asynccontextmanager
async def lifespan(my_app: FastAPI):
    """The lifespan function."""
    try:
        # Connect to the databse
        db_info = f"'{env['PGUSER']}@{env['PGHOST']}:{env['PGPORT']}'"
        while True:
            try:
                await connect_to_db(my_app, add_write_connection_pool=with_transactions)
                logger.info("Reached %r database on %s", env["PGDATABASE"], db_info)
                break
            except ConnectionRefusedError:
                logger.warning("Trying to reach %r database on %s", env["PGDATABASE"], db_info)

                # timeout gestion if specified
                if my_app.state.pg_timeout is not None:
                    my_app.state.pg_timeout -= my_app.state.pg_pause
                    if my_app.state.pg_timeout < 0:
                        sys.exit("Unable to start up catalog service")
                await asyncio.sleep(my_app.state.pg_pause)

        common_settings.set_http_client(httpx.AsyncClient(timeout=DEFAULT_TIMEOUT_CONFIG))

        # Run the data lifecycle management as an automatic periodic task
        lifecycle.run()

        yield

    finally:
        await lifecycle.cancel()
        await close_db_connection(my_app)
        await common_settings.del_http_client()

must_be_authenticated(route_path)

Return true if a user must be authenticated to use this endpoint route path.

Source code in docs/rs-server/services/catalog/rs_server_catalog/app.py

def must_be_authenticated(route_path: str) -> bool:
    """Return true if a user must be authenticated to use this endpoint route path."""

    # Remove the /catalog prefix, if any
    path = route_path.removeprefix(CATALOG_PREFIX)

    no_auth = path in TECH_ENDPOINTS or path.startswith("/auth/")
    return not no_auth