Need help with implementing "personalized" stac catalogues...

[parabelboi]

Hi @ALL,

I need some (high-level) advise on how to implement "personalized" stac catalogues with stac-fastapi-pgstac:

Requirement:
Each user has it's own isolated catalog (with enabled transactions extension).
The user should be extracted from a jwt token, that is provided by some authentication server (keycloak).

As the number of users is limited (i.e. less than 100), I'd say that each user could have it's own database.
One could setup one stac-fastapi-pgstac container per user and have an authentication proxy that dispatches to the per-user instances.
But that would multiply the database overhead per user (stored procedures, db-migration efforts etc).
And it would only work for a static set of users.

I assume the "correct" solution would be to extend the BasicCrudClient and implement it to "just" dispatch everything to dynamically instantiated (per-user) StacApi instances. This would at least allow to initialize the DB whenever a user is logging in for the first time.

Since I am new to python/fastapi/stac, I need some advice on how to solve this...

Many thanks,
parabelboi

[hrodmn]

Hi @parabelboi. Standing up a new database for each user sounds like a difficult path. I would work on building an authentication system in your stac-fastapi-pgstac application instead. There is a fair amount of material about adding authentication to FastAPI apps out there (e.g. https://fastapi.tiangolo.com/tutorial/security/)

It is early days, but there is also a new application called stac-auth-proxy which can be helpful for managing per-user STAC API access. Good luck! Please report back here with whatever system you come wind up using.

[parabelboi]

Hi @hrodmn,

thanks for pointing me to the authentication tutorial.

Regarding the stac_auth_proxy, it seems to be addressing a slightly different use case (not-fully-isolated).

I assume it would be possible (but with some effort) to setup the proxy with query-filter-rewriting-etc, so that all queries are only returning "private" collections or items, but it does not seem to be the optimal solution for fully isolated catalogues.

But I'll keep on researching, and report back as you suggested.

[parabelboi]

Hi @ALL,

I want to elaborate a little bit more on a solution for "isolated" (per-user) catalogs:

Could the following be a liable solution?

If I understand the parameter route_dependencies correctly, the following should be possible:

1. define and register a function that extracts the username from a JWT

2. define and register another function that creates a db connection for that user

   • i.e. using the original PostgresSetting, but with db name set to ${dbname}_${username})
   • to avoid re-opening the db connection for each request that connection should be stored somewhere, maybe a map

3. define and register a middleware that

   • sets up (or replaces) request.app.state.readpool and request.app.state.writepool with the per-user based database backends

4. all further processing (search, get_item, etc) would then be executed on the per-user backends
   -> problem solved

If I am making wrong assumptions, please let me know (python noob).

Cheers,
parabelboi

[parabelboi]

I just want to leave a comment with our current approach of implementing user-based cataloges.

Please keep in mind that I am new to python, so any hints are welcome 😁

app.py

We have implemented our app like this:

• no opening of the postgres database with the apps lifespan
• instead the database will be opened by the middleware/userdb.py within the lifecycle of the request

import os
import logging
import yaml

stac_api_id = os.getenv('STAC_API_ID', 'catalog-id')
stac_api_title = os.getenv('STAC_API_TITLE', 'catalog-title')
stac_api_desc = os.getenv('STAC_API_DESCRIPTION', 'catalog-desc')

from stac_fastapi.api.app import StacApi
from stac_fastapi.pgstac.app import (
    api,
    app,
    settings,
    ORJSONResponse,
    get_request_model,
    post_request_model,
)
from stac_fastapi.pgstac.app import application_extensions
from stac_fastapi.pgstac.core import CoreCrudClient
from stac_fastapi.extensions.core.transaction import TransactionExtension
from stac_fastapi.pgstac.transactions import BulkTransactionsClient, TransactionsClient
from fastapi import FastAPI
from starlette.middleware import Middleware
from middleware.userdb import PerUserDbMiddleware
from brotli_asgi import BrotliMiddleware
from stac_fastapi.api.middleware import CORSMiddleware, ProxyHeaderMiddleware

extensions = application_extensions
extensions.append(TransactionExtension(client=TransactionsClient(), settings=settings))

api = StacApi(
    app=FastAPI(
        openapi_url=settings.openapi_url,
        docs_url="/docs",
        redoc_url=None,
        root_path=settings.root_path,
        title=settings.stac_fastapi_title,
        version=settings.stac_fastapi_version,
        description=settings.stac_fastapi_description,
    ),
    title=stac_api_title,
    description=stac_api_desc,
    settings=settings,
    extensions=application_extensions,
    client=CoreCrudClient(pgstac_search_model=post_request_model),
    response_class=ORJSONResponse,
    search_get_request_model=get_request_model,
    search_post_request_model=post_request_model,
    middlewares=[
        Middleware(PerUserDbMiddleware),
        Middleware(JWTAuthMiddleware),
        Middleware(BrotliMiddleware),
        Middleware(ProxyHeaderMiddleware),
        Middleware(
            CORSMiddleware,
            allow_origins=settings.cors_origins,
            allow_methods=settings.cors_methods,
        ),
    ],
)

app = api.app

@app.on_event("startup")
async def startup_event():
    """Setup Logging"""
    with open('log_conf.yaml', 'r') as f:
        log_cfg = yaml.safe_load(f.read())
        logging.config.dictConfig(log_cfg)
    logger = logging.getLogger(__name__)
    logger.debug("Fasten your seat belt, we're starting up.")

middleware/userdb.py

The middleware expects the username to be set within the request by some sort of authentication layer (not covered here).

import logging
from starlette.types import ASGIApp, Receive, Scope, Send
from starlette.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from db import connect_to_db, close_db_connection

class PerUserDbMiddleware(BaseHTTPMiddleware):
    """Implement isolated (per-User) Database Connections

    If a valid user is found in the request, it will be used to setup a private database connection.
    """

    def __init__(self, app: ASGIApp):
        """Create the middleware."""
        super().__init__(app)

    async def dispatch(self, request, call_next):
        """Call from stac-fastapi framework"""

        logger = logging.getLogger(__name__)

        if not hasattr(request.state, "username"):
            return JSONResponse("Authentication failed for some reason.", status_code=403)

        username = request.state.username

        if username is None:
            return JSONResponse("You need to login to see your shopcart.", status_code=401)

        logger.debug("opening per-user db for user: %s", username)

        await connect_to_db(request, username)

        response = await call_next(request)

        await close_db_connection(request)
        return response

db.py

Our wrapper around the database is providing the following modifications to the original implementation:

• as soon as a user logs in for the first time, a cloned database will be created for him
• the users database will be named like this: $(dbname)_$(username)
• the database connection will be stored at the request.state instead of app.state
• a modified get_connection function will return this connection to the pgstac implementation
• not covered: caching db connections, so every request will open/close a db connection

"""Per-User Database connection handling."""

import logging

from stac_fastapi.pgstac.db import close_db_connection
from fastapi import FastAPI, Request
from typing import (
        AsyncIterator,
        Optional,
        Callable,
        Literal,
)
from stac_fastapi.pgstac.config import PostgresSettings
from stac_fastapi.pgstac.db import DB, translate_pgstac_errors

import asyncpg
from asyncpg import Connection, InvalidCatalogNameError
from contextlib import asynccontextmanager
import copy

async def close_db_connection(request: Request) -> None:
    """Close connection."""
    await request.state.readpool.close()
    await request.state.writepool.close()

@asynccontextmanager
async def get_connection(
    request: Request,
    readwrite: Literal["r", "w"] = "r",
) -> AsyncIterator[Connection]:
    """Retrieve connection from database conection pool."""
    pool = request.state.writepool if readwrite == "w" else request.state.readpool
    with translate_pgstac_errors():
        async with pool.acquire() as conn:
            yield conn

# TODO: have a proper cache with deleter functions that close lingering connections

async def clone_reader_db(
        app: FastAPI,
        username: str,
        postgres_settings: PostgresSettings,
) -> None:
    """Create a cloned database."""

    logger = logging.getLogger(__name__)
    logger.debug("clone_reader_db: creating database for user %s from template %s",
            username, postgres_settings.reader_connection_string)

    template = postgres_settings.postgres_dbname
    dbname = template + "_" + username
    dbuser = postgres_settings.postgres_user

    db_conn = await asyncpg.connect(postgres_settings.reader_connection_string)
    await db_conn.execute(
        f'CREATE DATABASE "{dbname}" WITH TEMPLATE "{template}" OWNER "{dbuser}"'
    )
    await db_conn.close()

async def connect_to_db(
        request: Request,
        username: Optional[str] = None,
        postgres_settings: Optional[PostgresSettings] = None,
) -> None:
    """Create connection pools & connection retriever on application."""

    if not postgres_settings:
        postgres_settings = PostgresSettings()

    logger = logging.getLogger(__name__)

    db = DB()

    if not username:
        logger.debug("connecting to default readpool: %s", postgres_settings.reader_connection_string)
        readpool = postgres_settings.reader_connection_string
        app.state.readpool = await db.create_pool(readpool, postgres_settings)
        return

    user_pg_settings = copy.copy(postgres_settings)
    user_pg_settings.postgres_dbname += "_" + username

    readpool = user_pg_settings.reader_connection_string
    writepool = user_pg_settings.writer_connection_string

    logger.debug("user_pg_settings.reader_connection_string: %s", readpool)
    logger.debug("user_pg_settings.writer_connection_string: %s", writepool)

    request.app.state.get_connection = get_connection

    try:
        request.state.readpool = await db.create_pool(readpool, user_pg_settings)
        request.state.writepool = await db.create_pool(writepool, user_pg_settings)
    except InvalidCatalogNameError as ex:
        # create the cloned db
        await clone_reader_db(request, username, postgres_settings=postgres_settings)
        # and retry the conection
        request.state.readpool = await db.create_pool(readpool, user_pg_settings)
        request.state.writepool = await db.create_pool(writepool, user_pg_settings)