API Documentation

This is a set of modest utilities that may be helpful when talking to PostgreSQL.

normanpg.errors

Sometimes things go wrong.

exception normanpg.errors.NormanPgException(message: str, inner: Exception = None)

Bases: Exception

This is the base exception for other exceptions defined in this library.

property inner

Get the exception that caused this exception.

property message

Get the exception message.

normanpg.functions.database

This module contains database-level functions.

class normanpg.functions.database.TempSchema(url: str, rand: int = 8, prefix: str = None)

Bases: object

This is a context manager you can use to create a temporary schema.

property prefix

Get the prefix.

property schema_name

Get the schema name.

property url

Get the database URL.

normanpg.functions.database.create_db(url: str, dbname: str, admindb: str = 'postgres')

Create a database on a Postgres instance.

Parameters

• url – the database URL

• dbname – the name of the database

• admindb – the name of an existing (presumably the main) database

normanpg.functions.database.create_extension(url: str, extension: str, dbname: str = None)

Create (install) an extension in a database.

Parameters

• url – the database URL

• extension – the name of an existing (presumably the main) database

• dbname – the name of the database

normanpg.functions.database.create_schema(url: str, schema: str, dbname: str = None)

Create a schema in the database.

Parameters

• url – the database URL

• schema – the name of the schema

• dbname – the name of the database

normanpg.functions.database.db_exists(url: str, dbname: str = None, admindb: str = 'postgres') → bool

Does a given database on a Postgres instance exist?

Parameters

• url – the database URL

• dbname – the name of the database to test

• admindb – the name of an existing (presumably the main) database

Returns

True if the database exists, otherwise False

normanpg.functions.database.drop_schema(url: str, schema: str, dbname: str = None, cascade: bool = True)

Drop a schema in the database.

Parameters

• url – the database URL

• schema – the name of the schema

• dbname – the name of the database

• cascade – True to drop the schema even if it is not empty, otherwise the attempt fails

normanpg.functions.database.parse_dbname(url: str) → str

Parse the database name from a connection URL.

Parameters

url – the URL

Returns

the database name

normanpg.functions.database.schema_exists(url: str, schema: str = None, dbname: str = None) → bool

Does a given schema exist within a database?

Parameters

• url – the database URL

• schema – the name of the database to test

• dbname – the name of the database

Returns

True if the schema exists, otherwise False

normanpg.functions.database.temp_name(rand: int = 8, prefix: str = None)

Generate a randomized of a specified length and an optional prefix.

Parameters

• rand – the number of random characters in the name

• prefix – the prefix

Returns

the randomized name

normanpg.functions.database.touch_db(url: str, dbname: str = None, admindb: str = 'postgres')

Create a database if it does not already exist.

Parameters

• url – the database URL

• dbname – the name of the database

• admindb – the name of an existing (presumably the main) database

normanpg.functions.tables

This module contains table-level functions.

exception normanpg.functions.tables.InvalidSrsException(message: str, inner: Exception = None)

Bases: normanpg.errors.NormanPgException

Raised if an attempt is made to reference an invalid spatial reference system (SRS).

exception normanpg.functions.tables.NoGeometryColumn(message: str, inner: Exception = None)

Bases: normanpg.errors.NormanPgException

Raised if an attempt is made to access non-existent geometries.

exception normanpg.functions.tables.TooManyGeometryColumns(message: str, inner: Exception = None)

Bases: normanpg.errors.NormanPgException

Raised if an attempt is made to access a table with multiple geometry columns.

normanpg.functions.tables.geometry_column(cnx: Union[str, psycopg2.extensions.connection], table_name: str, schema_name: str) → str

Get the name of the geometry column in a feature table.

Parameters

• cnx – an open connection or database connection string

• table_name – the name of the table

• schema_name –

Returns

the name of the geometry column

normanpg.functions.tables.srid(cnx: Union[str, psycopg2.extensions.connection], table_name: str, schema_name: str) → int

Get the SRID for geometries in a feature table.

Parameters

• cnx – an open connection or database connection string

• table_name – the name of the table

• schema_name – the name of the schema in which the table resides

Returns

the SRID for geometries in the table

normanpg.functions.tables.table_exists(cnx: Union[str, psycopg2.extensions.connection], table_name: str, schema_name: str) → bool

Parameters

• cnx – an open connection or database connection string

• table_name – the name of the table

• schema_name – the name of the schema in which the table resides

Returns

True if the table exists, otherwise False

normanpg.geometry

This module contains conveniences for working with geometries.

normanpg.geometry.shape(obj: str) → shapely.geometry.base.BaseGeometry

Convert a geometry from Postgres into a Shapely geometry.

Parameters

obj – the raw geometry retrieved from Postgres (a WKB hex string)

Returns

the Shapely geometry

normanpg.pg

This module needs a description.

normanpg.pg.DEFAULT_ADMIN_DB = 'postgres'

the default administrative database name

normanpg.pg.DEFAULT_PG_PORT = 5432

the default Postgres database port

exception normanpg.pg.InvalidDbResult(message: str, inner: Exception = None)

Bases: normanpg.errors.NormanPgException

Raised in response to an invalid result returned from the database.

normanpg.pg.compose_table(table_name: str, schema_name: str = None) → psycopg2.sql.Composed

Get a composed SQL object for a fully-qualified table name.

Parameters

• table_name – the table name

• schema_name – the schema name

Returns

a composed SQL object

normanpg.pg.connect(url: str, dbname: str = None, autocommit: bool = False) → psycopg2.extensions.connection

Get a connection to a Postgres database instance.

Parameters

• url – the instance URL

• dbname – the target database name

• autocommit – Set the autocommit flag on the connection?

Returns

a psycopg2 connection

Note

If the caller does not provide the dbname parameter the function creates a connection to the database specified in the URL.

normanpg.pg.execute(cnx: Union[str, psycopg2.extensions.connection], query: Union[str, psycopg2.sql.Composed], caller: str = None)

Execute a query that returns no result.

Parameters

• cnx – an open connection or database connection string

• query – the psycopg2 composed query

• caller – identifies the caller (for diagnostics)

See also

• execute_scalar()

• execute_rows()

normanpg.pg.execute_rows(cnx: Union[str, psycopg2.extensions.connection], query: Union[str, psycopg2.sql.Composed], caller: str = None) → Iterable[psycopg2.extras.DictRow]

Execute a query that returns an iteration of rows.

Parameters

• cnx – an open connection or database connection string

• query – the psycopg2 composed query

• caller – identifies the caller (for diagnostics)

Returns

an iteration of DictRow instances representing the row

normanpg.pg.execute_scalar(cnx: Union[str, psycopg2.extensions.connection], query: Union[str, psycopg2.sql.Composed], caller: str = None) → Any

Execute a query that returns a single, scalar result.

Parameters

• cnx – an open psycopg2 connection or the database URL

• query – the psycopg2 composed query

• caller – identifies the caller (for diagnostics)

Returns

the scalar string result (or None if the query returns no result)

normanpg.pg.log_query(crs: psycopg2.extensions.cursor, caller: str, query: str)

Log a SQL query.

Parameters

• crs – the execution cursor

• caller – the caller

• query – the query