privex.db.postgres

This module holds PostgresWrapper - a somewhat higher level class for interacting with PostgreSQL.

Copyright:

    +===================================================+
    |                 © 2019 Privex Inc.                |
    |               https://www.privex.io               |
    +===================================================+
    |                                                   |
    |        Privex's Python Database Library           |
    |        License: X11 / MIT                         |
    |                                                   |
    |        Originally Developed by Privex Inc.        |
    |        Core Developer(s):                         |
    |                                                   |
    |          (+)  Chris (@someguy123) [Privex]        |
    |                                                   |
    +===================================================+

Copyright (c) 2019     Privex Inc.   ( https://www.privex.io )

Classes

PostgresWrapper([db, db_user, db_host, db_pass])

Lightweight wrapper class for interacting with PostgreSQL databases.
class privex.db.postgres.PostgresWrapper(db=None, db_user='root', db_host=None, db_pass=None, **kwargs)[source]

Lightweight wrapper class for interacting with PostgreSQL databases.

Usage

class MyManager(PostgresWrapper):
    SCHEMAS: List[Tuple[str, str]] = [
        ('users', "CREATE TABLE users (id SERIAL PRIMARY KEY, name VARCHAR(50));"),
        ('items', "CREATE TABLE items (id INTEGER PRIMARY KEY, name VARCHAR(50));"),
    ]

    def get_items(self):
        return self.fetchall("SELECT * FROM items;")

    def find_item(self, id: int):
        return self.fetchone("SELECT * FROM items WHERE id = %s;", [id]);
property conn

Get or create a Postgres connection

db = None

PostgreSQL database name

drop_table(table: str) → bool[source]

Drop the table table if it exists. If the table exists, it will be dropped and True will be returned.

If the table doesn’t exist (thus can’t be dropped), False will be returned.

insert(_table: str, _cursor: psycopg2.extensions.cursor = None, **fields) → Union[privex.helpers.collections.DictObject, psycopg2.extensions.cursor][source]

Builds and executes an insert query into the table _table using the keyword arguments for column names and values.

>>> db = GenericDBWrapper(db='SomeDB')
>>> cur = db.insert('users', first_name='John', last_name='Doe', phone='+1-800-123-4567')
>>> cur.lastrowid
15
Parameters

  • _table (str) – The table to insert into

  • _cursor (GenericCursor) – Optionally, specify a cursor to use, instead of the default cursor

  • fields – Keyword args mapping column names to values

Return DictObject cur

If no custom cursor was specified, the cursor used to execute the query is converted into a DictObject before closing it, then the dict is returned.

Return GenericCursor cur

If a custom cursor (_cursor) was specified, then that cursor will NOT be auto-closed, and the original cursor instance will be returned.

last_insert_id(table_name: str, pk_name='id')[source]

Get the last inserted ID for a given table + primary key.

Example:

>>> db = PostgresWrapper(db='my_db')
>>> db.action('INSERT INTO users (first_name, last_name) VALUES (?, ?);', ['John', 'Doe'])
>>> last_id = db.last_insert_id('users')
>>> db.fetchone('SELECT first_name, last_name FROM users WHERE id = ?', [last_id])
Record(id=16, first_name='John', last_name='Doe')
Parameters

  • table_name (str) – The table you want the last insertion for

  • pk_name (str) – The primary key name, e.g. id, username etc.

Return Any last_id

The last pk_name inserted into table_name

list_tables(schema: str = None) → List[str][source]

Get a list of tables present in the current database.

Example:

>>> GenericDBWrapper().list_tables()
['sqlite_sequence', 'nodes', 'node_api', 'node_failures']
Return List[str] tables

A list of tables in the database

query(sql: str, *params, fetch='all', **kwparams) → Tuple[Optional[iter], Any, psycopg2.extensions.cursor][source]

Create an instance of your database wrapper:

>>> db = GenericDBWrapper()

Querying with prepared SQL queries and returning a single row:

>>> res, res_exec, cur = db.query("SELECT * FROM users WHERE first_name = ?;", ['John'], fetch='one')
>>> res
(12, 'John', 'Doe', '123 Example Road',)
>>> cur.close()

Querying with plain SQL queries, using query_mode, handling an iterator result, and using the cursor

If your database API returns rows as tuple``s or ``list``s, you can use ``query_mode='dict' (or set query_mode in the constructor) to convert any row results into dictionaries which map each column to their values.

>>> res, _, cur = db.query("SELECT * FROM users;", fetch='all', query_mode='dict')

When querying with fetch='all', depending on your database API, res may be an iterator, and cannot be accessed via an index like res[0].

You should make sure to iterate the rows using a for loop:

>>> for row in res:
...     print(row['first_name'], ':', row)
John : {'first_name': 'John', 'last_name': 'Doe', 'id': 12}
Dave : {'first_name': 'Dave', 'last_name': 'Johnson', 'id': 13}
Aaron : {'first_name': 'Aaron', 'last_name': 'Swartz', 'id': 14}

Or, if the result object is a generator, then you can auto-iterate the results into a list using x = list(res):

>>> rows = list(res)
>>> rows[0]
{'first_name': 'John', 'last_name': 'Doe', 'id': 12}

Using the returned cursor (third return item), we can access various metadata about our query. Note that cursor objects vary between database APIs, and not all methods/attributes may be available, or may return different data than shown below:

>>> cur.description  # cursor.description often contains the column names matching the query columns
(('id', None, None, None, None, None, None), ('first_name', None, None, None, None, None, None),
 ('last_name', None, None, None, None, None, None))

>>> _, _, cur = db.query("INSERT INTO users (first_name, last_name) VALUES ('a', 'b')", fetch='no')
>>> cur.rowcount   # cursor.rowcount tells us how many rows were affected by a query
1
>>> cur.lastrowid  # cursor.lastrowid tells us the ID of the last row we inserted with this cursor
3
Parameters

  • sql (str) – An SQL query to execute

  • params – Any positional arguments other than sql will be passed to cursor.execute.

  • fetch (str) – Fetch mode. Either all (return cursor.fetchall() as first return arg), one (return cursor.fetchone()), or no (do not fetch. first return arg is None).

  • kwparams – Any keyword arguments that aren’t specified as parameters / keyword args for this method will be forwarded to cursor.execute

Key GenericCursor cursor

Use this specific cursor instead of automatically obtaining one

Key cursor_name

If your database API supports named cursors (e.g. PostgreSQL), then you may specify cursor_name as a keyword argument to use a named cursor for this query.

Key query_mode

Either flat (fetch results as they were originally returned from the DB), or dict (use _zip_cols() to convert tuple/list rows into dicts mapping col:value).

Return iter results

(tuple item 1) An iterable such as a generator, or storage type e.g. list or dict. NOTE: If you’ve set fetch='all', depending on your database adapter, this may be a generator or other form of iterator that cannot be directly accessed via index (i.e. res[123]). Instead you must iterate it with a for loop, or cast it into a list/tuple to automatically iterate it into an indexed object, e.g. ``list(res)`

Return Any res_exec

(tuple item 2) The object returned from running cur.execute(sql, *params, **kwparams). This may be a cursor, but may also vary based on database API.

Return GenericCursor cur

(tuple item 3) The cursor that was used to execute and fetch your query. To allow for use with server side cursors, the cursor is NOT closed automatically. To avoid stale cursors, it’s best to run cur.close() when you’re done with handling the returned results.

table_exists(table: str, schema: str = None) → bool[source]

Returns True if the table table exists in the database, otherwise False.

>>> GenericDBWrapper().table_exists('some_table')
True
>>> GenericDBWrapper().table_exists('other_table')
False
Parameters

table (str) – The table to check for existence.

Return bool exists

True if the table table exists in the database, otherwise False.