Module stac_fastapi.pgstac.db¶

Database connection handling.

Variables¶

ConnectionGetter

Functions¶

close_db_connection¶

def close_db_connection(
    app: fastapi.applications.FastAPI
) -> None

Close connection.

con_init¶

def con_init(
    conn
)

Use orjson for json returns.

connect_to_db¶

def connect_to_db(
    app: fastapi.applications.FastAPI,
    get_conn: Optional[Callable[[starlette.requests.Request, Literal['r', 'w']], AsyncIterator[asyncpg.connection.Connection]]] = None
) -> None

Create connection pools & connection retriever on application.

dbfunc¶

def dbfunc(
    conn: asyncpg.connection.Connection,
    func: str,
    arg: Union[str, Dict, List]
)

Wrap PLPGSQL Functions.

Keyword arguments: pool -- the asyncpg pool to use to connect to the database func -- the name of the PostgreSQL function to call arg -- the argument to the PostgreSQL function as either a string or a dict that will be converted into jsonb

get_connection¶

def get_connection(
    request: starlette.requests.Request,
    readwrite: Literal['r', 'w'] = 'r'
) -> AsyncIterator[asyncpg.connection.Connection]

Retrieve connection from database conection pool.

translate_pgstac_errors¶

def translate_pgstac_errors(

) -> Generator[NoneType, NoneType, NoneType]

Context manager that translates pgstac errors into FastAPI errors.

Classes¶

DB¶

class DB(
    connection_string=None,
    pool=None,
    connection=None
)

DB class that can be used with context manager.

Class variables¶

connection_string

Methods¶

create_pool¶

def create_pool(
    self,
    connection_string: str,
    settings
)

Create a connection pool.