cheesefactory-database

A wrapper for psycopg2.

Main Features

• Built on psycopg2.
• Pandas dataframe support.
• Test table existence.
• Test field existence.

Note: This package is still in beta status. As such, future versions may not be backwards compatible and features may change.

Installation

The source is hosted at https://bitbucket.org/hellsgrannies/cheesefactory-database

pip install cheesefactory-database

Dependencies

• psycopg2

License

Examples

Connect to the remote PostgreSQL server and execute a query

from cheesefactory_database.postgresql import CfPostgresql

db = CfPostgresql(host='myhostname', user='myusername', password='mypassword', database='mydatabase')

Args:

• host (str): PostgreSQL server hostname/IP. Default: 127.0.0.1
• port (str): PostgreSQL server port. Default: 5432
• user (str): Username for authentication.
• password (str): Password for authentication.
• database (str): Database for connection
• autocommit (bool): Use autocommit on changes? Default: False
• dictionary_cursor (bool): Return the results as a dictionary? Default: False
• encoding (str): Database client encoding ("utf8", "latin1", "usascii"). Default: utf8

Returns:

• (CfPostgresql): An instance of the database connection.

Execute a query

results = db.execute("SELECT first_name, last_name FROM person WHERE last_name = 'Smith'")

Args:

• query (str): SQL query to execute.
• dataframe (bool): Output the results to a pandas.Dataframe?
• fetchall (bool): Perform a fetchall() and return the results?

Returns:

• If fetchall=False, None is returned.
• If dataframe=True, a pandas.Dataframe is returned.

Check connection status

status = db.connection_status()

Returns:

• (str): Connection status: "OK" or "Error"

Check database existence

status = db.database_exists(database_name='my_database')

Args:

• database_name (str): Name of the database to search for

Returns:

• (bool): True, if database exists. False, if not.

Check schema existence

status = db.schema_exists(schema_name='my_schema')

Args:

• schema_name (str): Name of the schema to search for.

Returns:

• (bool): True, if schema exists. False, if not.

Check table existence

status = db.table_exists(table_path='my_schema.my_table')

Args:

• table_path (str): Table name in the form <schema>.<table>

Returns:

• (bool): True, if table exists. False, if not.

Check field existence

status = db.fields_exist(table_path='my_schema.my_table', table_fields=['first_name', 'last_name'])

Args:

• table_path (str): Table name in the form <schema>.<table>
• table_fields (str): A list of fields to check.

Returns:

• (bool): True, if all fields are present in the table. False, if not.

Get a table's primary keys

primary_keys = db.get_primary_keys(table_path='my_schema.my_table')

Args:

• table_path (str): Table name in the form <schema>.<table>

Returns:

• (List): A list of primary keys.

Quote PostgreSQL reserved words in a list

quoted_word_list = CfPostgresql.quote_reserved_words(word_list)

Args:

• word_list (List): A list of words to check

Returnes:

• (List): word_list with all PostgreSQL reserved words surrounded in double-quotes.