Facata
Python DBAPI simplified
Install / Use
/learn @tlocke/FacataREADME
= Facata :toc: preamble :toclevels: 1
A Python library that provides a simplified alternative to https://www.python.org/dev/peps/pep-0249/[DBAPI 2]. It provides a facade in front of DBAPI 2 drivers.
== 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 https://www.sqlite.org/[SQLite] database. It uses the https://docs.python.org/3/library/sqlite3.html[sqlite3] driver that comes with Python:
[source,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 https://en.wikipedia.org/wiki/Database_transaction#In_SQL[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:
[source,python]
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:
[source,python]
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:
[source,python]
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:
[source,python]
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:
[source,python]
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 https://www.sqlite.org/lang_keywords.html[surround it
by double quotes]. But you can't do:
[source,python]
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 https://docs.python.org/3/tutorial/introduction.html#strings[triple quotes] to delimit the string instead:
[source,python]
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:
[source,python]
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:
[source,python]
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 https://www.postgresql.org/[PostgreSQL] with the
https://github.com/tlocke/pg8000[pg8000] driver requires the dependency pg8000 to be
installed by doing pip install facata[pg8000]. Then connect to a database as follows:
[source,python]
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:
[source,python]
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 https://www.postgresql.org/docs/current/tutorial-transactions.html[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.r
