Python DBAPI simplified

Overview

Facata

Installation

  • Create a virtual environment: python3 -m venv venv

  • Activate the virtual environment: source venv/bin/activate

  • Install: pip install facata

Examples For SQLite With SQLite3

Basic Example

Here’s an example of how to access an SQLite database. It uses the sqlite3 driver that comes with Python:

>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3", dbname=":memory:")
>>>
>>> # Create a temporary table
>>>
>>> con.run("CREATE TEMPORARY TABLE book (title TEXT)")
>>>
>>> # Populate the table
>>>
>>> for title in ("Ender's Game", "The Magus"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> # Print all the rows in the table
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
("Ender's Game",)
('The Magus',)

Transactions

Here’s how to run groups of SQL statements in a transaction:

>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3", dbname=":memory:")
>>>
>>> con.run("BEGIN TRANSACTION")
>>> con.run("CREATE TEMPORARY TABLE book (title TEXT)")
>>> for title in ("Ender's Game", "The Magus", "Phineas Finn"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>> con.run("COMMIT")
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
("Ender's Game",)
('The Magus',)
('Phineas Finn',)

Rolling back a transaction:

>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3", dbname=":memory:")
>>>
>>> con.run("BEGIN TRANSACTION")
>>> con.run("CREATE TEMPORARY TABLE book (title TEXT)")
>>> for title in ("Ender's Game", "The Magus", "Phineas Finn"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>> con.run("COMMIT")
>>>
>>> con.run("BEGIN TRANSACTION")
>>> con.run("DELETE FROM book WHERE title = :title", title="Phineas Finn")
>>> con.run("ROLLBACK")
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
("Ender's Game",)
('The Magus',)
('Phineas Finn',)

Query Using Fuctions

Another query, using an SQLite function:

>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3",  dbname=":memory:")
>>>
>>> con.run("SELECT datetime(1092941466, 'unixepoch');")
[('2004-08-19 18:51:06',)]

Retrieve Column Metadata From Results

Find the column metadata returned from a query:

>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3", dbname=":memory:")
>>>
>>> con.run("create temporary table quark (spin text)")
>>> for spin in ('Up', 'Down'):
...     con.run("INSERT INTO quark (spin) VALUES (:spin)", spin=spin)
>>> # Now execute the query
>>>
>>> con.run("SELECT * FROM quark")
[('Up',), ('Down',)]
>>>
>>> # and read the metadata
>>>
>>> con.columns
[{'name': 'spin', 'type_code': None, 'display_size': None, 'internal_size': None, 'precision': None, 'scale': None}]
>>>
>>> # Show just the column names
>>>
>>> [c['name'] for c in con.columns]
['spin']

Many SQL Statements Can’t Be Parameterized

In SQLite parameters can only be used for data values. Sometimes this might not work as expected, for example the following fails:

>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3", dbname=":memory:")
>>>
>>> con.run("SELECT 'silo 1' LIMIT :lim", lim='ALL')
Traceback (most recent call last):
sqlite3.IntegrityError: datatype mismatch

You might think that the following would work, but in fact the server doesn’t like it:

>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3", dbname=":memory:")
>>>
>>> con.run("SELECT 'silo 1' WHERE 'a' IN :v", v=('a', 'b'))
Traceback (most recent call last):
sqlite3.OperationalError: near ":v": syntax error

Execute SQL Scripts

If you want to execute a series of SQL statements (eg. an .sql file), with SQLite you need to access the the underlying connection with and use SQLite3’s executescript() method:

>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3", dbname=":memory:")
>>>
>>> statements = """
...     CREATE TEMPORARY TABLE quark (spin text);
...     INSERT INTO quark (spin) VALUES ('Up');
...     INSERT INTO quark (spin) VALUES ('Down');"""
>>>
>>> con.connection.executescript(statements)
<sqlite3.Cursor object at ...>
>>> con.run("SELECT * FROM quark")
[('Up',), ('Down',)]

A caveat is that when executing scripts you can’t have any parameters.

Quoted Identifiers in SQL

Say you had a column called My Column. Since it’s case sensitive and contains a space, you’d have to surround it by double quotes. But you can’t do:

>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3", dbname=":memory:")
>>>
>>> con.run("select 'hello' as "My Column"")
Traceback (most recent call last):
SyntaxError: invalid syntax

since Python uses double quotes to delimit string literals, so one solution is to use Python’s triple quotes to delimit the string instead:

>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3", dbname=":memory:")
>>>
>>> con.run('''select 'hello' as "My Column"''')
[('hello',)]

Custom Adapter From A Python Type To An SQLite Type

Sqlite3 has a mapping from Python types to PostgreSQL types for when it needs to send SQL parameters to the server. The default mapping that comes with SQLite is fairly limited, but you can add custom conversions with an adapter.

By default, a Python decimal.Decimal object can’t be used as a parameter but here’s an example of how to register an adapter for it:

>>> from decimal import Decimal
>>> import facata
>>>
>>> con = facata.connect("sqlite", "sqlite3", dbname=":memory:")
>>>
>>> def decimal_py_to_db(dec):
...     return str(dec)  # Must return int, float, str or bytes
>>>
>>> con.register_py_to_db(Decimal, None, decimal_py_to_db)
>>>
>>> con.run("SELECT :val", val=Decimal('0.1'))
[('0.1',)]

Note that it still came back as a str object because we only changed the mapping from Python to SQLite. See below for an example of how to change the mapping from SQLite to Python.

Custom Adapter From An SQLite Type To A Python Type

SQLite3 has a mapping from SQLite types to Python types for when it receives SQL results from the server. With a custom adapter you can add a new mapping or replace the default mapping. Here’s an example:

>>> from decimal import Decimal
>>> import sqlite3
>>> import facata
>>>
>>> con = facata.connect(
...     "sqlite", "sqlite3", dbname=":memory:", detect_types=sqlite3.PARSE_DECLTYPES)
>>>
>>> def decimal_db_to_py(data):  # The parameter is of type bytes
...     return Decimal(data.decode('ascii'))
>>>
>>> con.register_db_to_py('decimal', decimal_db_to_py)
>>>
>>> con.run("CREATE TEMPORARY TABLE book (title TEXT, price decimal)")
>>> con.run(
...     "INSERT INTO book (title, price) VALUES (:title, :price)",
...     title="The Island", price='7.99')
>>>
>>> con.run("SELECT * FROM book")
[('The Island', Decimal('7.99'))]

Note that registering the 'db to py' adapter only afected the mapping from the SQLite type to the Python type. See above for an example of how to change the mapping from Python to SQLite.

Examples For PostgreSQL With pg8000

Connecting To A Database

Connecting to PostgreSQL with the pg8000 driver requires the dependency pg8000 to be installed by doing pip install facata[pg8000]. Then connect to a database as follows:

>>> import facata
>>>
>>> with facata.connect("postgresql", "pg8000", user="postgres", password="pw") as con:
...     con.run("SELECT 'Hello'")
[['Hello']]

Basic Example

Import facacta, connect to the database, create a table, add some rows and then query the table:

>>> import facata
>>>
>>> # Connect to the database with user name postgres
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> # Create a temporary table
>>>
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> # Populate the table
>>>
>>> for title in ("Ender's Game", "The Magus"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> # Print all the rows in the table
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
[1, "Ender's Game"]
[2, 'The Magus']

Transactions

Here’s how to run groups of SQL statements in a transaction:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("START TRANSACTION")
>>>
>>> # Create a temporary table
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> for title in ("Ender's Game", "The Magus", "Phineas Finn"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> con.run("COMMIT")
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
[1, "Ender's Game"]
[2, 'The Magus']
[3, 'Phineas Finn']

rolling back a transaction:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> # Create a temporary table
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> for title in ("Ender's Game", "The Magus", "Phineas Finn"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> con.run("START TRANSACTION")
>>>
>>> con.run("DELETE FROM book WHERE title = :title", title="Phineas Finn")
>>>
>>> con.run("ROLLBACK")
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
[1, "Ender's Game"]
[2, 'The Magus']
[3, 'Phineas Finn']

Query Using Fuctions

Another query, using some PostgreSQL functions:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("SELECT extract(millennium from now())")
[[3.0]]

Interval Type

A query that returns the PostgreSQL interval type:

>>> import datetime
>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> ts = datetime.date(1980, 4, 27)
>>> con.run("SELECT timestamp '2013-12-01 16:06' - :ts", ts=ts)
[[datetime.timedelta(days=12271, seconds=57960)]]

Point Type

A round-trip with a PostgreSQL point type:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("SELECT CAST(:pt as point)", pt='(2.3,1)')
[['(2.3,1)']]

Client Encoding

When communicating with the server, pg8000 uses the character set that the server asks it to use (the client encoding). By default the client encoding is the database’s character set (chosen when the database is created), but the client encoding can be changed in a number of ways (eg. setting CLIENT_ENCODING in postgresql.conf). Another way of changing the client encoding is by using an SQL command. For example:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("SET CLIENT_ENCODING TO 'UTF8'")
>>>
>>> con.run("SHOW CLIENT_ENCODING")
[['UTF8']]

JSON

JSON always comes back from the server de-serialized. If the JSON you want to send is a dict then you can just do:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> val = {'name': 'Apollo 11 Cave', 'zebra': True, 'age': 26.003}
>>> con.run("SELECT :apollo", apollo=val)
[[{'age': 26.003, 'name': 'Apollo 11 Cave', 'zebra': True}]]

JSON can always be sent in serialized form to the server:

>>> import json
>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> val = ['Apollo 11 Cave', True, 26.003]
>>> con.run("SELECT CAST(:apollo as jsonb)", apollo=json.dumps(val))
[[['Apollo 11 Cave', True, 26.003]]]

Retrieve Column Metadata From Results

Find the column metadata returned from a query:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("create temporary table quark (id serial, name text)")
>>>
>>> for name in ('Up', 'Down'):
...     con.run("INSERT INTO quark (name) VALUES (:name)", name=name)
>>>
>>> # Now execute the query
>>>
>>> con.run("SELECT * FROM quark")
[[1, 'Up'], [2, 'Down']]
>>>
>>> # and retried the metadata
>>>
>>> con.columns
[{'table_oid': ..., 'column_attrnum': 1, 'type_oid': 23, 'type_size': 4, 'type_modifier': -1, 'format': 0, 'name': 'id'}, {'table_oid': ..., 'column_attrnum': 2, 'type_oid': 25, 'type_size': -1, 'type_modifier': -1, 'format': 0, 'name': 'name'}]
>>>
>>> # Show just the column names
>>>
>>> [c['name'] for c in con.columns]
['id', 'name']

Notices, Notifications And Parameter Statuses

To access the PostgreSQL notices you need to obtain the underlying pg8000 connection. The notices are stored in a deque called Connection.notices and added using the append() method. Similarly there are Connection.notifications for notifications and Connection.parameter_statuses for changes to the server configuration. Here’s an example:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("LISTEN aliens_landed")
>>> con.run("NOTIFY aliens_landed")
>>>
>>> # A notification is a tuple containing (backend_pid, channel, payload)
>>>
>>> con.connection.notifications[0]
(..., 'aliens_landed', '')

LIMIT ALL

You might think that the following would work, but in fact it fails:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("SELECT 'silo 1' LIMIT :lim", lim='ALL')
Traceback (most recent call last):
pg8000.exceptions.DatabaseError: ...

Instead the docs say that you can send null as an alternative to ALL, which does work:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("SELECT 'silo 1' LIMIT :lim", lim=None)
[['silo 1']]

IN and NOT IN

You might think that the following would work, but in fact the server doesn’t like it:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("SELECT 'silo 1' WHERE 'a' IN :v", v=('a', 'b'))
Traceback (most recent call last):
pg8000.exceptions.DatabaseError: ...

instead you can write it using the unnest function:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("SELECT 'silo 1' WHERE 'a' IN (SELECT unnest(:v))", v=('a', 'b'))
[['silo 1']]

and you can do the same for NOT IN.

Many SQL Statements Can’t Be Parameterized

In PostgreSQL parameters can only be used for data values, not identifiers. Sometimes this might not work as expected, for example the following fails:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("CREATE USER juan WITH PASSWORD :password", password='quail')
Traceback (most recent call last):
pg8000.exceptions.DatabaseError: ...

It fails because the PostgreSQL server doesn’t allow this statement to have any parameters. There are many SQL statements that one might think would have parameters, but don’t.

COPY From And To A File

The SQL COPY statement can be used to copy from and to a file or file-like object. First we need to obtain the underlying pg8000 connection. Here' an example using the CSV format:

>>> from io import StringIO
>>> import csv
>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>> connection = con.connection
>>>
>>> # Create a CSV file in memory
>>>
>>> stream_in = StringIO()
>>> csv_writer = csv.writer(stream_in)
>>> csv_writer.writerow([1, "electron"])
12
>>> csv_writer.writerow([2, "muon"])
8
>>> csv_writer.writerow([3, "tau"])
7
>>> stream_in.seek(0)
0
>>>
>>> # Create a table and then copy the CSV into it
>>>
>>> connection.run("CREATE TEMPORARY TABLE lepton (id SERIAL, name TEXT)")
>>>
>>> connection.run("COPY lepton FROM STDIN WITH (FORMAT CSV)", stream=stream_in)
>>>
>>> # COPY from a table to a stream
>>>
>>> stream_out = StringIO()
>>> connection.run("COPY lepton TO STDOUT WITH (FORMAT CSV)", stream=stream_out)
>>> stream_out.seek(0)
0
>>> for row in csv.reader(stream_out):
...     print(row)
['1', 'electron']
['2', 'muon']
['3', 'tau']

Execute Multiple SQL Statements

If you want to execute a series of SQL statements (eg. an .sql file), you can run them as expected:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> statements = "SELECT 5; SELECT 'Erich Fromm';"
>>>
>>> con.run(statements)
[[5], ['Erich Fromm']]

The only caveat is that when executing multiple statements you can’t have any parameters.

Quoted Identifiers in SQL

Say you had a column called My Column. Since it’s case sensitive and contains a space, you’d have to surround it by double quotes. But you can’t do:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("select 'hello' as "My Column"")
Traceback (most recent call last):
SyntaxError: invalid syntax

since Python uses double quotes to delimit string literals, so one solution is to use Python’s triple quotes to delimit the string instead:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run('''select 'hello' as "My Column"''')
[['hello']]

Custom Adapter From A Python Type To A PostgreSQL Type

pg8000 has a mapping from Python types to PostgreSQL types for when it needs to send SQL parameters to the server. The default mapping that comes with pg8000 is designed to work well in most cases, but you might want to add or replace the default mapping.

A Python datetime.timedelta object is sent to the server as a PostgreSQL interval type, which has the oid 1186. But let’s say we wanted to create our own Python class to be sent as an interval type. Then we’d have to register an adapter:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> class MyInterval(str):
...     pass
>>>
>>> def my_interval_py_to_db(my_interval):
...     return my_interval  # Must return a str
>>>
>>> con.register_py_to_db(MyInterval, 1186, my_interval_py_to_db)
>>> con.run("SELECT :interval", interval=MyInterval("2 hours"))
[[datetime.timedelta(seconds=7200)]]

Note that it still came back as a datetime.timedelta object because we only changed the mapping from Python to PostgreSQL. See below for an example of how to change the mapping from PostgreSQL to Python.

Custom adapter from a PostgreSQL type to a Python type

pg8000 has a mapping from PostgreSQL types to Python types for when it receives SQL results from the server. The default mapping that comes with pg8000 is designed to work well in most cases, but you might want to add or replace the default mapping.

If pg800 recieves PostgreSQL interval type, which has the oid 1186, it converts it into a Python datetime.timedelta object. But let’s say we wanted to create our own Python class to be used instead of datetime.timedelta. Then we’d have to register an adapter:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> class MyInterval(str):
...     pass
>>>
>>> def my_interval_db_to_py(my_interval_str):  # The parameter is of type str
...     return MyInterval(my_interval)
>>>
>>> con.register_db_to_py(1186, my_interval_db_to_py)
>>> con.run("SELECT \'2 years'")
[['2 years']]

Note that registering the 'in' adapter only afects the mapping from the PostgreSQL type to the Python type. See above for an example of how to change the mapping from PostgreSQL to Python.

Could Not Determine Data Type Of Parameter

Sometimes you’ll get the 'could not determine data type of parameter' error message from the server:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("SELECT :v IS NULL", v=None)
Traceback (most recent call last):
pg8000.exceptions.DatabaseError: {'S': 'ERROR', 'V': 'ERROR', 'C': '42P18', 'M': 'could not determine data type of parameter $1', 'F': 'postgres.c', 'L': '...', 'R': 'exec_parse_message'}

One way of solving it is to put a cast in the SQL:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("SELECT cast(:v as TIMESTAMP) IS NULL", v=None)
[[True]]

Another way is to override the type that pg8000 sends along with each parameter:

>>> import facata
>>> import pg8000.native
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("SELECT :v IS NULL", v=None, types={'v': pg8000.native.TIMESTAMP})
[[True]]

Prepared Statements

Prepared statements can be useful in improving performance when you have a statement that’s executed repeatedly. Here’s an example:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> # Create the prepared statement
>>> ps = con.prepare("SELECT cast(:v as varchar)")
>>>
>>> # Exceute the statement repeatedly
>>> ps.run(v="speedy")
[['speedy']]
>>> ps.run(v="rapid")
[['rapid']]
>>> ps.run(v="swift")
[['swift']]
>>>
>>> # Close the prepared statement, releasing resources on the server
>>> ps.close()

Use Environment Variables As Connection Defaults

You might want to use the current user as the database username for example:

import getpass
import facata

# Connect to the database with current user name
user = getpass.getuser()
conection = facata.connect("postgresql", "pg8000", user=user, password="pw")

connection.run("SELECT 'pilau'")

or perhaps you may want to use some of the same environment variables that libpq uses:

>>> import facata
>>> from os import environ
>>>
>>> user = environ.get('PGUSER', 'postgres')
>>> password = environ.get('PGPASSWORD', 'pw')
>>> host = environ.get('PGHOST', 'localhost')
>>> port = environ.get('PGPORT', '5432')
>>> dbname = environ.get('PGDATABASE')
>>>
>>> conection = facata.connect(
...    "postgresql", "pg8000", user=user, password=password, host=host, port=port,
...    dbname=dbname)
>>>
>>> connection.run("SELECT 'Mr Cairo'")
[['Mr Cairo']]

It might be asked, why doesn’t Facata have this behaviour built in? The thinking follows the second aphorism of The Zen of Python:

Explicit is better than implicit.

So we’ve taken the approach of only being able to set connection parameters using the facata.connect() function.

Connect To PostgreSQL Over SSL

To connect to the server using SSL defaults do:

from facata import connect


connection = connect(
    "postgresql", "pg8000", user="postgres", password="cpsnow", ssl_context=True)
connection.run("SELECT 'The game is afoot!'")

To connect over SSL with custom settings, set the ssl_context parameter to an ssl.SSLContext object:

from facata import connect
import ssl


ssl_context = ssl.SSLContext()
ssl_context.verify_mode = ssl.CERT_REQUIRED
ssl_context.load_verify_locations('root.pem')
connection = connect(
    "postgresql", "pg8000", user="postgres", password="cpsnow", ssl_context=ssl_context)

It may be that your PostgreSQL server is behind an SSL proxy server in which case you can set a pg8000-specific attribute ssl.SSLContext.request_ssl = False which tells pg8000 to connect using an SSL socket, but not to request SSL from the PostgreSQL server:

from facata import connect
import ssl


ssl_context = ssl.SSLContext()
ssl_context.request_ssl = False
connection = connect(
    "postgresql", "pg8000", user="postgres", password="cpsnow", ssl_context=ssl_context)

Server-Side Cursors

You can use the SQL commands DECLARE, FETCH, MOVE and CLOSE to manipulate server-side cursors. For example:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> con.run("START TRANSACTION")
>>> con.run("DECLARE c SCROLL CURSOR FOR SELECT * FROM generate_series(1, 100)")
>>> con.run("FETCH FORWARD 5 FROM c")
[[1], [2], [3], [4], [5]]
>>>
>>> con.run("MOVE FORWARD 50 FROM c")
>>> con.run("FETCH BACKWARD 10 FROM c")
[[54], [53], [52], [51], [50], [49], [48], [47], [46], [45]]
>>>
>>> con.run("CLOSE c")
>>> con.run("ROLLBACK")

BLOBs (Binary Large Objects)

There’s a set of SQL functions for manipulating BLOBs. Here’s an example:

>>> import facata
>>>
>>> con = facata.connect("postgresql", "pg8000", user="postgres", password="pw")
>>>
>>> # Create a BLOB and get its oid
>>> data = b'hello'
>>> res = con.run("SELECT lo_from_bytea(0, :data)", data=data)
>>> oid = res[0][0]
>>>
>>> # Create a table and store the oid of the BLOB
>>> con.run("CREATE TEMPORARY TABLE image (raster oid)")
>>> con.run("INSERT INTO image (raster) VALUES (:oid)", oid=oid)
>>> # Retrieve the data using the oid
>>> con.run("SELECT lo_get(:oid)", oid=oid)
[[b'hello']]
>>>
>>> # Add some data to the end of the BLOB
>>> more_data = b' all'
>>> offset = len(data)
>>> con.run(
...     "SELECT lo_put(:oid, :offset, :data)",
...     oid=oid, offset=offset, data=more_data)
[['']]
>>> con.run("SELECT lo_get(:oid)", oid=oid)
[[b'hello all']]
>>>
>>> # Download a part of the data
>>> con.run("SELECT lo_get(:oid, 6, 3)", oid=oid)
[[b'all']]

Examples For PostgreSQL With Psycopg2

Connecting To A Database

Connecting to PostgreSQL with the psycopg2 driver requires the dependency psycopg2 to be installed by doing pip install facata[psycopg2]. This will install the binary version, but for production use see the psycopg installation docs. Then connect to a database as follows:

>>> import facata
>>>
>>> with facata.connect(
...         "postgresql", "psycopg2", user="postgres", password="pw",
...         host="localhost") as con:
...     con.run("SELECT 'Hello'")
[('Hello',)]

Basic Example

Import facacta, connect to the database, create a table, add some rows and then query the table:

>>> import facata
>>>
>>> # Connect to the database with user name postgres
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> # Create a temporary table
>>>
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> # Populate the table
>>>
>>> for title in ("Ender's Game", "The Magus"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> # Print all the rows in the table
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
(1, "Ender's Game")
(2, 'The Magus')

Transactions

Here’s how to run groups of SQL statements in a transaction:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run("START TRANSACTION")
>>>
>>> # Create a temporary table
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> for title in ("Ender's Game", "The Magus", "Phineas Finn"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> con.run("COMMIT")
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
(1, "Ender's Game")
(2, 'The Magus')
(3, 'Phineas Finn')

rolling back a transaction:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> # Create a temporary table
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> for title in ("Ender's Game", "The Magus", "Phineas Finn"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> con.run("START TRANSACTION")
>>>
>>> con.run("DELETE FROM book WHERE title = :title", title="Phineas Finn")
>>>
>>> con.run("ROLLBACK")
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
(1, "Ender's Game")
(2, 'The Magus')
(3, 'Phineas Finn')

Query Using Fuctions

Another query, using some PostgreSQL functions:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run("SELECT extract(millennium from now())")
[(3.0,)]

Interval Type

A query that returns the PostgreSQL interval type:

>>> import datetime
>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> ts = datetime.date(1980, 4, 27)
>>> con.run("SELECT timestamp '2013-12-01 16:06' - :ts", ts=ts)
[(datetime.timedelta(days=12271, seconds=57960),)]

Point Type

A round-trip with a PostgreSQL point type:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run("SELECT CAST(:pt as point)", pt='(2.3,1)')
[('(2.3,1)',)]

Client Encoding

When communicating with the server, psycopg2 uses the character set that the server asks it to use (the client encoding). By default the client encoding is the database’s character set (chosen when the database is created), but the client encoding can be changed in a number of ways (eg. setting CLIENT_ENCODING in postgresql.conf). Another way of changing the client encoding is by using an SQL command. For example:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run("SET CLIENT_ENCODING TO 'UTF8'")
>>>
>>> con.run("SHOW CLIENT_ENCODING")
[('UTF8',)]

Retrieve Column Metadata From Results

Find the column metadata returned from a query:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run("create temporary table quark (id serial, name text)")
>>>
>>> for name in ('Up', 'Down'):
...     con.run("INSERT INTO quark (name) VALUES (:name)", name=name)
>>>
>>> # Now execute the query
>>>
>>> con.run("SELECT * FROM quark")
[(1, 'Up'), (2, 'Down')]
>>>
>>> # and retried the metadata
>>>
>>> con.columns
[{'name': 'id', 'type_code': 23, 'display_size': None, 'internal_size': 4, 'precision': None, 'scale': None}, {'name': 'name', 'type_code': 25, 'display_size': None, 'internal_size': -1, 'precision': None, 'scale': None}]
>>>
>>> # Show just the column names
>>>
>>> [c['name'] for c in con.columns]
['id', 'name']

Notices, Notifications And Parameter Statuses

To access the PostgreSQL notices you need to obtain the underlying pg8000 connection. The notices are stored in a deque called Connection.notices and added using the append() method. Similarly there are Connection.notifications for notifications and Connection.parameter_statuses for changes to the server configuration. Here’s an example:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run("LISTEN aliens_landed")
>>> con.run("NOTIFY aliens_landed")
>>>
>>> # A notification is a tuple containing (backend_pid, channel, payload)
>>>
>>> connection = con.connection
>>> connection.notifies[0]
Notify(..., 'aliens_landed', '')

LIMIT ALL

You might think that the following would work, but in fact it fails:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run("SELECT 'silo 1' LIMIT :lim", lim='ALL')
Traceback (most recent call last):
psycopg2.errors.InvalidTextRepresentation: ...

Instead the docs say that you can send null as an alternative to ALL, which does work:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run("SELECT 'silo 1' LIMIT :lim", lim=None)
[('silo 1',)]

COPY From And To A File

The SQL COPY statement can be used to copy from and to a file or file-like object. First we need to obtain the underlying Psycopg2 connection. Here' an example using the CSV format:

>>> from io import StringIO
>>> import csv
>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> # Create a CSV file in memory
>>>
>>> stream_in = StringIO()
>>> csv_writer = csv.writer(stream_in)
>>> csv_writer.writerow([1, "electron"])
12
>>> csv_writer.writerow([2, "muon"])
8
>>> csv_writer.writerow([3, "tau"])
7
>>> stream_in.seek(0)
0
>>>
>>> # Create a table and then copy the CSV into it
>>>
>>> con.run("CREATE TEMPORARY TABLE lepton (id SERIAL, name TEXT)")
>>>
>>> cur = con.connection.cursor()
>>> cur.copy_expert("COPY lepton FROM STDIN WITH (FORMAT CSV)", stream_in)
>>>
>>> # COPY from a table to a stream
>>>
>>> stream_out = StringIO()
>>> cur.copy_expert("COPY lepton TO STDOUT WITH (FORMAT CSV)", stream_out)
>>> stream_out.seek(0)
0
>>> for row in csv.reader(stream_out):
...     print(row)
['1', 'electron']
['2', 'muon']
['3', 'tau']

Execute Multiple SQL Statements

If you want to execute a series of SQL statements (eg. an .sql file), you can run them as expected:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> statements = "SELECT 5; SELECT 'Erich Fromm';"
>>>
>>> con.run(statements)
[('Erich Fromm',)]

The only caveat is that when executing multiple statements you can’t have any parameters.

Quoted Identifiers in SQL

Say you had a column called My Column. Since it’s case sensitive and contains a space, you’d have to surround it by double quotes. But you can’t do:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run("select 'hello' as "My Column"")
Traceback (most recent call last):
SyntaxError: invalid syntax

since Python uses double quotes to delimit string literals, so one solution is to use Python’s triple quotes to delimit the string instead:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run('''select 'hello' as "My Column"''')
[('hello',)]

Custom Adapter From A Python Type To A PostgreSQL Type

Psycopg2 has a mapping from Python types to PostgreSQL types for when it needs to send SQL parameters to the server. The default mapping that comes with Psycopg2 is designed to work well in most cases, but you might want to add or replace the default mapping.

A Python datetime.timedelta object is sent to the server as a PostgreSQL interval type, which has the oid 1186. But let’s say we wanted to create our own Python class to be sent as an interval type. Then we’d have to register an adapter:

>>> import facata
>>> from psycopg2.extensions import AsIs
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> class MyInterval(str):
...     pass
>>>
>>> def my_interval_py_to_db(my_interval):
...     return AsIs(f"'{my_interval}'")  # Must return a str
>>>
>>> con.register_py_to_db(MyInterval, None, my_interval_py_to_db)
>>> con.run("SELECT cast(:interval as interval)", interval=MyInterval("2 hours"))
[(datetime.timedelta(seconds=7200),)]

Note that it still came back as a datetime.timedelta object because we only changed the mapping from Python to PostgreSQL. See below for an example of how to change the mapping from PostgreSQL to Python.

Custom adapter from a PostgreSQL type to a Python type

Psycopg2 has a mapping from PostgreSQL types to Python types for when it receives SQL results from the server. The default mapping that comes with Psycopg2 is designed to work well in most cases, but you might want to add or replace the default mapping.

If Psycopg2 recieves PostgreSQL interval type, which has the oid 1186, it converts it into a Python datetime.timedelta object. But let’s say we wanted to create our own Python class to be used instead of datetime.timedelta. Then we’d have to register an adapter:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> class MyInterval(str):
...     pass
>>>
>>> def my_interval_db_to_py(my_interval_str, cur):  # The parameter is of type str
...     return MyInterval(my_interval_str)
>>>
>>> con.register_db_to_py(1186, my_interval_db_to_py)
>>> con.run("SELECT cast('2 years' as interval)")
[('2 years',)]

Note that registering the 'in' adapter only afects the mapping from the PostgreSQL type to the Python type. See above for an example of how to change the mapping from PostgreSQL to Python.

Prepared Statements

Prepared statements aren’t currently supported by Psycopg2, so the Facata just defaults to standard execution behind the scenes:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> # Create the prepared statement
>>> ps = con.prepare("SELECT cast(:v as varchar)")
>>>
>>> # Exceute the statement repeatedly
>>> ps.run(v="speedy")
[('speedy',)]
>>> ps.run(v="rapid")
[('rapid',)]
>>> ps.run(v="swift")
[('swift',)]
>>>
>>> # Close the prepared statement, releasing resources on the server
>>> ps.close()

Use Environment Variables As Connection Defaults

You might want to use the current user as the database username for example:

import getpass
import facata

# Connect to the database with current user name
user = getpass.getuser()
con = facata.connect(
    "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")

con.run("SELECT 'pilau'")

or perhaps you may want to use some of the same environment variables that libpq uses:

>>> import facata
>>> from os import environ
>>>
>>> user = environ.get('PGUSER', 'postgres')
>>> password = environ.get('PGPASSWORD', 'pw')
>>> host = environ.get('PGHOST', 'localhost')
>>> port = environ.get('PGPORT', '5432')
>>> dbname = environ.get('PGDATABASE')
>>>
>>> connection = facata.connect(
...    "postgresql", "psycopg2", user=user, password=password, host=host, port=port,
...    dbname=dbname)
>>>
>>> connection.run("SELECT 'Mr Cairo'")
[('Mr Cairo',)]

It might be asked, why doesn’t Facata have this behaviour built in? The thinking follows the second aphorism of The Zen of Python:

Explicit is better than implicit.

So we’ve taken the approach of only being able to set connection parameters using the facata.connect() function.

Connect To PostgreSQL Over SSL

To connect to the server using SSL defaults do:

from facata import connect


connection = connect(
    "postgresql", "psycopg2", user="postgres", password="cpsnow", sslmode='require')
connection.run("SELECT 'The game is afoot!'")

To connect over SSL with custom settings, use the libpq parameters:

from facata import connect
import ssl


connection = connect(
    "postgresql", "psycopg2", user="postgres", password="cpsnow",
    sslmode="verify-full", sslrootcert="root.pem")

Server-Side Cursors

You can use the SQL commands DECLARE, FETCH, MOVE and CLOSE to manipulate server-side cursors. For example:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> con.run("START TRANSACTION")
>>> con.run("DECLARE c SCROLL CURSOR FOR SELECT * FROM generate_series(1, 100)")
>>> con.run("FETCH FORWARD 5 FROM c")
[(1,), (2,), (3,), (4,), (5,)]
>>>
>>> con.run("MOVE FORWARD 50 FROM c")
>>> con.run("FETCH BACKWARD 10 FROM c")
[(54,), (53,), (52,), (51,), (50,), (49,), (48,), (47,), (46,), (45,)]
>>>
>>> con.run("CLOSE c")
>>> con.run("ROLLBACK")

BLOBs (Binary Large Objects)

There’s a set of SQL functions for manipulating BLOBs. Here’s an example:

>>> import facata
>>>
>>> con = facata.connect(
...     "postgresql", "psycopg2", user="postgres", password="pw", host="localhost")
>>>
>>> # Create a BLOB and get its oid
>>> data = b'hello'
>>> res = con.run("SELECT lo_from_bytea(0, :data)", data=data)
>>> oid = res[0][0]
>>>
>>> # Create a table and store the oid of the BLOB
>>> con.run("CREATE TEMPORARY TABLE image (raster oid)")
>>> con.run("INSERT INTO image (raster) VALUES (:oid)", oid=oid)
>>> # Retrieve the data using the oid
>>> result = con.run("SELECT lo_get(:oid)", oid=oid)
>>> bytes(result[0][0])
b'hello'
>>>
>>> # Add some data to the end of the BLOB
>>> more_data = b' all'
>>> offset = len(data)
>>> con.run(
...     "SELECT lo_put(:oid, :offset, :data)",
...     oid=oid, offset=offset, data=more_data)
[('',)]
>>> result = con.run("SELECT lo_get(:oid)", oid=oid)
>>> bytes(result[0][0])
b'hello all'
>>>
>>> # Download a part of the data
>>> result = con.run("SELECT lo_get(:oid, 6, 3)", oid=oid)
>>> bytes(result[0][0])
b'all'

Examples For MariaDB With MariaDb Driver

Connecting To A Database

Connecting to MariaDB with the mariadb driver requires the Connector/C to be installed, and then the dependency mariadb can be installed by doing pip install facata[mariadb].

>>> import facata
>>>
>>> with facata.connect(
...         "mariadb", "mariadb", user="root", password="pw", host="127.0.0.1",
...         port=3306, dbname="mysql") as con:
...     con.run("SELECT 'Hello'")
[('Hello',)]

Basic Example

Connect to the database, create a table, add some rows and then query the table:

>>> import facata
>>>
>>> con = facata.connect(
...     "mariadb", "mariadb", user="root", password="pw", host="127.0.0.1", port=3306,
...     dbname="mysql")
>>>
>>> # Create a temporary table
>>>
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> # Populate the table
>>>
>>> for title in ("Ender's Game", "The Magus"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> # Print all the rows in the table
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
(1, "Ender's Game")
(2, 'The Magus')

Transactions

Here’s how to run groups of SQL statements in a transaction:

>>> import facata
>>>
>>> con = facata.connect(
...     "mariadb", "mariadb", user="root", password="pw", host="127.0.0.1", port=3306,
...     dbname="mysql")
>>>
>>> con.run("START TRANSACTION")
>>>
>>> # Create a temporary table
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> for title in ("Ender's Game", "The Magus", "Phineas Finn"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>> con.run("COMMIT")
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
(1, "Ender's Game")
(2, 'The Magus')
(3, 'Phineas Finn')

rolling back a transaction:

>>> import facata
>>>
>>> con = facata.connect(
...     "mariadb", "mariadb", user="root", password="pw", host="127.0.0.1", port=3306,
...     dbname="mysql")
>>>
>>> # Create a temporary table
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> for title in ("Ender's Game", "The Magus", "Phineas Finn"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> con.run("START TRANSACTION")
>>> con.run("DELETE FROM book WHERE title = :title", title="Phineas Finn")
>>> con.run("ROLLBACK")
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
(1, "Ender's Game")
(2, 'The Magus')
(3, 'Phineas Finn')

Query Using Fuctions

Another query, using a MariaDB function:

>>> import facata
>>>
>>> con = facata.connect(
...     "mariadb", "mariadb", user="root", password="pw", host="127.0.0.1", port=3306)
>>>
>>> con.run("SELECT GREATEST(0, -3, 2)")
[(2,)]

Point Type

A round-trip with a MariaDB point type:

>>> import facata
>>>
>>> con = facata.connect(
...     "mariadb", "mariadb", user="root", password="pw", host="127.0.0.1", port=3306,
...     dbname="mysql")
>>>
>>> con.run("SELECT AsText(PointFromText(:pt))", pt='POINT(2.3 1)')
[('POINT(2.3 1)',)]

Retrieve Column Metadata From Results

Find the column metadata returned from a query:

>>> import facata
>>>
>>> con = facata.connect(
...     "mariadb", "mariadb", user="root", password="pw", host="127.0.0.1", port=3306,
...     dbname="mysql")
>>>
>>> con.run("create temporary table quark (id serial, name text)")
>>> for name in ('Up', 'Down'):
...     con.run("INSERT INTO quark (name) VALUES (:name)", name=name)
>>> # Now execute the query
>>>
>>> con.run("SELECT * FROM quark")
[(1, 'Up'), (2, 'Down')]
>>>
>>> # and retried the metadata
>>>
>>> con.columns
[{'name': 'id', 'type_code': 8, 'display_size': 5, 'internal_size': 20, 'precision': 0, 'scale': 0}, {'name': 'name', 'type_code': 252, 'display_size': 65535, 'internal_size': 262140, 'precision': 0, 'scale': 0}]
>>>
>>> # Show just the column names
>>>
>>> [c['name'] for c in con.columns]
['id', 'name']

Load Data From A File

The SQL LOAD DATA INFILE statement can be used to copy from a file. Here' an example:

>>> import facata
>>>
>>> con = facata.connect(
...     "mariadb", "mariadb", user="root", password="pw", host="127.0.0.1", port=3306,
...     dbname="mysql")
>>>
>>> # Open and display a pre-prepared file
>>>
>>> with open('test/mariadb/data.dat') as f:
...     f.read()
'\telectron\n\tmuon\n\ttau\n'
>>>
>>> # Create a table and populate it
>>>
>>> con.run("CREATE TEMPORARY TABLE lepton (id SERIAL, name TEXT)")
>>> con.run("LOAD DATA LOCAL INFILE 'test/mariadb/data.dat' INTO TABLE lepton")
>>>
>>> con.run("SELECT * FROM lepton")
[(1, 'electron'), (2, 'muon'), (3, 'tau')]

Quoted Identifiers in SQL

Say you had a column called My Column. Since it’s case sensitive and contains a space, you’d have to surround it by backticks. But you can’t do:

>>> import facata
>>>
>>> con = facata.connect(
...     "mariadb", "mariadb", user="root", password="pw", host="127.0.0.1", port=3306)
>>>
>>> con.run("select 'hello' as `My Column`")
[('hello',)]

Connect To MariaDB Over SSL

To connect to the server using SSL do something like:

import facata

con = facata.connect(
    "mariadb", "mariadb", user="root", password="pw", host="127.0.0.1", port=3306,
    ssl_ca='ca-cert.pem', ssl_cert='client-cert.pem', ssl_key='client-key.pem')

con.run("SELECT 'The game is afoot!'")

Examples For MySQL With MySQL Connector/Python

Connecting to a database

Connecting to MySQL with the MySQL Connector driver requires the dependency mysql-connector to be installed by doing pip install facata[mysql-connector]. Then connect to a database as follows:

>>> import facata
>>>
>>> with facata.connect(
...         "mysql", "mysql-connector", user="root", password="pw", dbname="mysql",
...         host="127.0.0.1", port=3305) as con:
...     con.run("SELECT 'Hello'")
[('Hello',)]

Basic Example

Connect to the database, create a table, add some rows and then query the table:

>>> import facata
>>>
>>> con = facata.connect(
...     "mysql", "mysql-connector", user="root", password="pw", host="127.0.0.1",
...     port=3305, dbname="mysql")
>>>
>>> # Create a temporary table
>>>
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> # Populate the table
>>>
>>> for title in ("Ender's Game", "The Magus"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> # Print all the rows in the table
>>>
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
(1, "Ender's Game")
(2, 'The Magus')

Transactions

Here’s how to run groups of SQL statements in a transaction:

>>> import facata
>>>
>>> con = facata.connect(
...     "mysql", "mysql-connector", user="root", password="pw", host="127.0.0.1",
...     port=3305, dbname="mysql")
>>>
>>> con.run("START TRANSACTION")
>>>
>>> # Create a temporary table
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> for title in ("Ender's Game", "The Magus", "Phineas Finn"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>> con.run("COMMIT")
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
(1, "Ender's Game")
(2, 'The Magus')
(3, 'Phineas Finn')

rolling back a transaction:

>>> import facata
>>>
>>> con = facata.connect(
...     "mysql", "mysql-connector", user="root", password="pw", host="127.0.0.1",
...     port=3305, dbname="mysql")
>>>
>>> # Create a temporary table
>>> con.run("CREATE TEMPORARY TABLE book (id SERIAL, title TEXT)")
>>>
>>> for title in ("Ender's Game", "The Magus", "Phineas Finn"):
...     con.run("INSERT INTO book (title) VALUES (:title)", title=title)
>>>
>>> con.run("START TRANSACTION")
>>> con.run("DELETE FROM book WHERE title = :title", title="Phineas Finn")
>>> con.run("ROLLBACK")
>>> for row in con.run("SELECT * FROM book"):
...     print(row)
(1, "Ender's Game")
(2, 'The Magus')
(3, 'Phineas Finn')

Query Using Fuctions

Another query, using some MySQL functions:

>>> import facata
>>>
>>> con = facata.connect(
...     "mysql", "mysql-connector", user="root", password="pw", host="127.0.0.1",
...     port=3305, dbname="mysql")
>>>
>>> con.run("SELECT GREATEST(0, -3, 2)")
[(2,)]

Point Type

A round-trip with a MySQL point type:

>>> import facata
>>>
>>> con = facata.connect(
...     "mysql", "mysql-connector", user="root", password="pw", host="127.0.0.1",
...     port=3305, dbname="mysql")
>>>
>>> con.run("SELECT ST_AsText(ST_GeomFromText(:pt))", pt='POINT(2.3 1)')
[('POINT(2.3 1)',)]

Retrieve Column Metadata From Results

Find the column metadata returned from a query:

>>> import facata
>>>
>>> con = facata.connect(
...     "mysql", "mysql-connector", user="root", password="pw", host="127.0.0.1",
...     port=3305, dbname="mysql")
>>>
>>> con.run("create temporary table quark (id serial, name text)")
>>> for name in ('Up', 'Down'):
...     con.run("INSERT INTO quark (name) VALUES (:name)", name=name)
>>> # Now execute the query
>>>
>>> con.run("SELECT * FROM quark")
[(1, 'Up'), (2, 'Down')]
>>>
>>> # and retried the metadata
>>>
>>> con.columns
[{'name': 'id', 'type_code': 8, 'display_size': None, 'internal_size': None, 'precision': None, 'scale': None}, {'name': 'name', 'type_code': 252, 'display_size': None, 'internal_size': None, 'precision': None, 'scale': None}]
>>>
>>> # Show just the column names
>>>
>>> [c['name'] for c in con.columns]
['id', 'name']

Quoted Identifiers in SQL

Say you had a column called My Column. Since it’s case sensitive and contains a space, you’d have to surround it by backticks:

>>> import facata
>>>
>>> con = facata.connect(
...     "mysql", "mysql-connector", user="root", password="pw", host="127.0.0.1",
...     port=3305, dbname="mysql")
>>>
>>> con.run("select 'hello' as `My Column`")
[('hello',)]

Connect To MySQL Over SSL

To connect to the server using SSL do:

import facata
from mysql.connector.constants import ClientFlag


con = facata.connect(
    "mysql", "mysql-connector", user="root", password="pw", host="127.0.0.1",
    port=3305, dbname="mysql", client_flags=[ClientFlag.SSL],
    ssl_ca='/opt/mysql/ssl/ca.pem', ssl_cert='/opt/mysql/ssl/client-cert.pem',
    ssl_key='/opt/mysql/ssl/client-key.pem')

con.run("SELECT 'The game is afoot!'")

API Docs

facata.connect()

connect(dbms, driver, dbname=None, user=None, password=None, host=None, port=None, **params):: Returns a new facata.Connection representing a connection to the database. Connections aren’t threadsafe.

dbms

Name of DBMS. Can be mariadb, postgresql or sqlite.

driver

The name of a driver for the DBMS. The possible drivers for each DBMS are:

  • mariadb: (mariadb)

  • postgresql: (pg80000, psycopg2)

  • sqlite: (sqlite3)

dbname

Name of the database to connect to.

user

User name.

password

Password of the user.

host

Hostname

port

Port number.

params

Dictionary of parameters to be given as keyword arguments to the underlying driver.

facata.Connection

A facata.Connection object has the following methods and properties:

run(sql, **params)

Returns the result of executing the sql statement as a sequence of sequences (eg. a list of tuple`s) or `None if the statement isn’t one that can return results.

sql

The SQL statement as a str with any parameters written as a : followed by the parameter name.

params

Parameter name / value pairs.

Diferences From DBAPI 2

DBAPI 2 is the Python standard for accessing SQL databases. Facata differs from DBAPI 2 in these ways:

  • DBAPI 2 has five ways of writing parameters in SQL, but Facata has just one.

  • In DAPI 2 parameter values can be provided as a sequence or mapping, but in Facata parameter values are provided as keyword arguments.

  • DBAPI 2 has four levels of threadsafety, but Facata just has one.

  • DBAPI 2 has a 'cursor' object, but in Facata there is no cursor object and statements are executed using the connection.

  • DBAPI 2 overrides the autocommit mode of the underlying database by silently sending START TRANSACTION statements if a transaction isn’t already in progress. Facata never silently sends SQL statements.

Testing

  • Activate the virtual environment: source venv/bin/activate

  • Install tox: pip install tox

  • Run tox: tox

Doing A Release Of Facata

Run tox to make sure all tests pass, then update the release notes, then do:

git tag -a x.y.z -m "version x.y.z"
rm -r dist
python setup.py sdist bdist_wheel
for f in dist/*; do gpg --detach-sign -a $f; done
twine upload dist/*

Release Notes

Version 0.0.0, 2021-08-21

  • First release.

Owner
Tony Locke
When an inner situation is not made conscious, it appears outside as fate. - CG Jung
Tony Locke
A library for python made by me,to make the use of MySQL easier and more pythonic

my_ezql A library for python made by me,to make the use of MySQL easier and more pythonic This library was made by Tony Hasson , a 25 year old student

3 Nov 19, 2021
Redis Python Client

redis-py The Python interface to the Redis key-value store. Python 2 Compatibility Note redis-py 3.5.x will be the last version of redis-py that suppo

Andy McCurdy 11k Dec 29, 2022
DataStax Python Driver for Apache Cassandra

DataStax Driver for Apache Cassandra A modern, feature-rich and highly-tunable Python client library for Apache Cassandra (2.1+) and DataStax Enterpri

DataStax 1.3k Dec 25, 2022
Async ODM (Object Document Mapper) for MongoDB based on python type hints

ODMantic Documentation: https://art049.github.io/odmantic/ Asynchronous ODM(Object Document Mapper) for MongoDB based on standard python type hints. I

Arthur Pastel 732 Dec 31, 2022
A supercharged SQLite library for Python

SuperSQLite: a supercharged SQLite library for Python A feature-packed Python package and for utilizing SQLite in Python by Plasticity. It is intended

Plasticity 703 Dec 30, 2022
asyncio (PEP 3156) Redis support

aioredis asyncio (PEP 3156) Redis client library. Features hiredis parser Yes Pure-python parser Yes Low-level & High-level APIs Yes Connections Pool

aio-libs 2.2k Jan 04, 2023
A simple python package that perform SQL Server Source Control and Auto Deployment.

deploydb Deploy your database objects automatically when the git branch is updated. Production-ready! ⚙️ Easy-to-use 🔨 Customizable 🔧 Installation I

Mert Güvençli 10 Dec 07, 2022
Pysolr — Python Solr client

pysolr pysolr is a lightweight Python client for Apache Solr. It provides an interface that queries the server and returns results based on the query.

Haystack Search 626 Dec 01, 2022
PostgreSQL database adapter for the Python programming language

psycopg2 - Python-PostgreSQL Database Adapter Psycopg is the most popular PostgreSQL database adapter for the Python programming language. Its main fe

The Psycopg Team 2.8k Jan 05, 2023
edaSQL is a library to link SQL to Exploratory Data Analysis and further more in the Data Engineering.

edaSQL is a python library to bridge the SQL with Exploratory Data Analysis where you can connect to the Database and insert the queries. The query results can be passed to the EDA tool which can giv

Tamil Selvan 8 Dec 12, 2022
Py2neo is a comprehensive toolkit for working with Neo4j from within Python applications or from the command line.

Py2neo v3 Py2neo is a client library and toolkit for working with Neo4j from within Python applications and from the command line. The core library ha

64 Oct 14, 2022
Py2neo is a comprehensive toolkit for working with Neo4j from within Python applications or from the command line.

Py2neo Py2neo is a client library and toolkit for working with Neo4j from within Python applications and from the command line. The library supports b

Nigel Small 1.2k Jan 02, 2023
dbd is a database prototyping tool that enables data analysts and engineers to quickly load and transform data in SQL databases.

dbd: database prototyping tool dbd is a database prototyping tool that enables data analysts and engineers to quickly load and transform data in SQL d

Zdenek Svoboda 47 Dec 07, 2022
Python MYSQL CheatSheet.

Python MYSQL CheatSheet Python mysql cheatsheet. Install Required Windows(WAMP) Download and Install from HERE Linux(LAMP) install packages. sudo apt

Mohammad Dori 4 Jul 15, 2022
A Python Object-Document-Mapper for working with MongoDB

MongoEngine Info: MongoEngine is an ORM-like layer on top of PyMongo. Repository: https://github.com/MongoEngine/mongoengine Author: Harry Marr (http:

MongoEngine 3.9k Jan 08, 2023
GINO Is Not ORM - a Python asyncio ORM on SQLAlchemy core.

GINO - GINO Is Not ORM - is a lightweight asynchronous ORM built on top of SQLAlchemy core for Python asyncio. GINO 1.0 supports only PostgreSQL with

GINO Community 2.5k Dec 29, 2022
A Telegram Bot to manage Redis Database.

A Telegram Bot to manage Redis database. Direct deploy on heroku Manual Deployment python3, git is required Clone repo git clone https://github.com/bu

Amit Sharma 4 Oct 21, 2022
Script em python para carregar os arquivos de cnpj dos dados públicos da Receita Federal em MYSQL.

cnpj-mysql Script em python para carregar os arquivos de cnpj dos dados públicos da Receita Federal em MYSQL. Dados públicos de cnpj no site da Receit

17 Dec 25, 2022
Neo4j Bolt driver for Python

Neo4j Bolt Driver for Python This repository contains the official Neo4j driver for Python. Each driver release (from 4.0 upwards) is built specifical

Neo4j 762 Dec 30, 2022
SpyQL - SQL with Python in the middle

SpyQL SQL with Python in the middle Concept SpyQL is a query language that combines: the simplicity and structure of SQL with the power and readabilit

Daniel Moura 853 Dec 30, 2022