Skip to content

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:

Name Type Description
dict list[dict]

the new parameters list with the owner id parameter.
Source code in docs/rs-server/services/catalog/rs_server_catalog/app.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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

data_lifecycle(request) async

Trigger the data lifecycle management

Source code in docs/rs-server/services/catalog/rs_server_catalog/app.py 
158
159
160
161
@app.router.get("/data/lifecycle", include_in_schema=False)
async def data_lifecycle(request: Request):
    """Trigger the data lifecycle management"""
    await lifecycle.periodic_once(request)

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 
170
171
172
173
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 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@asynccontextmanager
async def lifespan(my_app: FastAPI):
    """The lifespan function."""
    try:
        # Connect to the databse
        db_info = f"'{env['POSTGRES_USER']}@{env['POSTGRES_HOST']}:{env['POSTGRES_PORT']}'"
        while True:
            try:
                await connect_to_db(my_app, add_write_connection_pool=with_transactions)
                logger.info("Reached %r database on %s", env["POSTGRES_DB"], db_info)
                break
            except ConnectionRefusedError:
                logger.warning("Trying to reach %r database on %s", env["POSTGRES_DB"], 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 
55
56
57
58
59
60
61
62
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