SkillAgentSearch skills...

Facata

Python DBAPI simplified

Install / Use

/learn @tlocke/Facata
About this skill

Quality Score

0/100

Supported Platforms

Universal

README

= 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

View on GitHub
GitHub Stars45
CategoryDevelopment
Updated3y ago
Forks2

Languages

Python

Security Score

75/100

Audited on Jan 28, 2023

No findings