2. SQLite

  • Database API in Python

2.1. SQL Syntax

Note

For SQL Syntax refer to SQL

2.1.1. Data Types

Table 2.7. SQLite data types

Type

Description

NULL

The value is a undefined value

INTEGER

The value is a signed integer, stored in 1, 2, 3, 4, 6, or 8 bytes depending on the magnitude of the value

REAL

The value is a floating point value, stored as an 8-byte IEEE floating point number

TEXT

The value is a text string, stored using the database encoding (UTF-8, UTF-16BE or UTF-16LE)

BLOB

The value is a blob of data, stored exactly as it was input

2.1.2. Constrains

Table 2.8. SQL Constraints

Constraint

Description

NOT NULL

Ensures that a column cannot have a NULL value

UNIQUE

Ensures that all values in a column are different

PRIMARY KEY

A combination of a NOT NULL and UNIQUE. Uniquely identifies each row in a table

FOREIGN KEY

Uniquely identifies a row/record in another table

CHECK

Ensures that all values in a column satisfies a specific condition

DEFAULT

Sets a default value for a column when no value is specified

INDEX

Used to create and retrieve data from the database very quickly

2.2. Connection

Listing 2.30. Connection to in-memory database
import sqlite3

with sqlite3.connect(':memory:') as db:
    db.execute('SELECT * FROM users')
Listing 2.31. Connection to database file
import sqlite3

with sqlite3.connect('database.sqlite3') as db:
    db.execute('SELECT * FROM users')

2.3. Execute

Listing 2.32. Execute
import sqlite3


SQL_CREATE_TABLE = """
    CREATE TABLE IF NOT EXISTS astronauts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        pesel INTEGER UNIQUE,
        firstname TEXT,
        lastname TEXT)"""
SQL_INSERT = 'INSERT INTO astronauts VALUES (NULL, :pesel, :firstname, :lastname)'

data = {'pesel': '61041212345', 'firstname': 'José', 'lastname': 'Jiménez'}


with sqlite3.connect(':memory:') as db:
    db.execute(SQL_CREATE_TABLE)

    try:
        db.execute(SQL_INSERT, data)
    except sqlite3.IntegrityError:
        print('Pesel need to be UNIQUE')

2.4. Executemany

2.4.1. List[tuple]

Listing 2.33. Execute many
import sqlite3


SQL_CREATE_TABLE = """
    CREATE TABLE IF NOT EXISTS astronauts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        pesel INTEGER UNIQUE,
        firstname TEXT,
        lastname TEXT)"""

SQL_INSERT_TUPLE = 'INSERT INTO astronauts VALUES (NULL, ?, ?, ?)'


list_of_tuples = [
    (61041212345, 'José', 'Jiménez'),
    (61041212346, 'Jan', 'Twardowski'),
    (61041212347, 'Melissa', 'Lewis'),
    (61041212348, 'Alex', 'Vogel'),
    (61041212349, 'Ryan', 'Stone'),
]


with sqlite3.connect(':memory:') as db:
    db.execute(SQL_CREATE_TABLE)

    try:
        db.executemany(SQL_INSERT_TUPLE, list_of_tuples)
    except sqlite3.IntegrityError:
        print('Pesel need to be UNIQUE')

2.4.2. List[dict]

Listing 2.34. Execute many
import sqlite3


SQL_CREATE_TABLE = """
    CREATE TABLE IF NOT EXISTS astronauts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        pesel INTEGER UNIQUE,
        firstname TEXT,
        lastname TEXT)"""

SQL_INSERT_DICT = 'INSERT INTO astronauts VALUES (NULL, :pesel, :firstname, :lastname)'


list_of_dicts = [
    {'pesel': '61041212345', 'firstname': 'José', 'lastname': 'Jiménez'},
    {'pesel': '61041212346', 'firstname': 'Jan', 'lastname': 'Twardowski'},
    {'pesel': '61041212347', 'firstname': 'Melissa', 'lastname': 'Lewis'},
    {'pesel': '61041212348', 'firstname': 'Alex', 'lastname': 'Vogel'},
    {'pesel': '61041212349', 'firstname': 'Ryan', 'lastname': 'Stone'},
]


with sqlite3.connect(':memory:') as db:
    db.execute(SQL_CREATE_TABLE)

    try:
        db.executemany(SQL_INSERT_DICT, list_of_dicts)
    except sqlite3.IntegrityError:
        print('Pesel need to be UNIQUE')

2.5. Results

2.5.1. Fetch as List[tuple]

Listing 2.35. Results
import sqlite3


SQL_CREATE_TABLE = """
    CREATE TABLE IF NOT EXISTS astronauts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        pesel INTEGER UNIQUE,
        firstname TEXT,
        lastname TEXT)"""
SQL_INSERT = 'INSERT INTO astronauts VALUES (NULL, :pesel, :firstname, :lastname)'
SQL_SELECT = 'SELECT * FROM astronauts'


astronauts = [
    {'pesel': '61041212345', 'firstname': 'José', 'lastname': 'Jiménez'},
    {'pesel': '61041212346', 'firstname': 'Jan', 'lastname': 'Twardowski'},
    {'pesel': '61041212347', 'firstname': 'Melissa', 'lastname': 'Lewis'},
    {'pesel': '61041212348', 'firstname': 'Alex', 'lastname': 'Vogel'},
    {'pesel': '61041212349', 'firstname': 'Ryan', 'lastname': 'Stone'},
]


with sqlite3.connect(':memory:') as db:
    db.execute(SQL_CREATE_TABLE)
    db.executemany(SQL_INSERT, astronauts)

    for row in db.execute(SQL_SELECT):
        print(row)

# (1, 61041212345, 'José', 'Jiménez')
# (2, 61041212346, 'Jan', 'Twardowski')
# (3, 61041212347, 'Melissa', 'Lewis')
# (4, 61041212348, 'Alex', 'Vogel')
# (5, 61041212349, 'Ryan', 'Stone')

2.5.2. Fetch as List[dict]

Listing 2.36. Results with dict
import sqlite3


SQL_CREATE_TABLE = """
    CREATE TABLE IF NOT EXISTS astronauts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        pesel INTEGER UNIQUE,
        firstname TEXT,
        lastname TEXT)"""
SQL_INSERT = 'INSERT INTO astronauts VALUES (NULL, :pesel, :firstname, :lastname)'
SQL_SELECT = 'SELECT * FROM astronauts'


astronauts = [
    {'pesel': '61041212345', 'firstname': 'José', 'lastname': 'Jiménez'},
    {'pesel': '61041212346', 'firstname': 'Jan', 'lastname': 'Twardowski'},
    {'pesel': '61041212347', 'firstname': 'Melissa', 'lastname': 'Lewis'},
    {'pesel': '61041212348', 'firstname': 'Alex', 'lastname': 'Vogel'},
    {'pesel': '61041212349', 'firstname': 'Ryan', 'lastname': 'Stone'},
]


with sqlite3.connect(':memory:') as db:
    db.execute(SQL_CREATE_TABLE)
    db.executemany(SQL_INSERT, astronauts)

    db.row_factory = sqlite3.Row

    for row in db.execute(SQL_SELECT):
        print(dict(row))

# {'id': 1, 'pesel': 61041212345, 'firstname': 'José', 'lastname': 'Jiménez'}
# {'id': 2, 'pesel': 61041212346, 'firstname': 'Jan', 'lastname': 'Twardowski'}
# {'id': 3, 'pesel': 61041212347, 'firstname': 'Melissa', 'lastname': 'Lewis'}
# {'id': 4, 'pesel': 61041212348, 'firstname': 'Alex', 'lastname': 'Vogel'}
# {'id': 5, 'pesel': 61041212349, 'firstname': 'Ryan', 'lastname': 'Stone'}

2.6. Cursor

Listing 2.37. Results with cursor
import sqlite3


SQL_CREATE_TABLE = """
    CREATE TABLE IF NOT EXISTS astronauts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        pesel INTEGER UNIQUE,
        firstname TEXT,
        lastname TEXT)"""
SQL_INSERT = 'INSERT INTO astronauts VALUES (NULL, :pesel, :firstname, :lastname)'
SQL_SELECT = 'SELECT * FROM astronauts'


astronauts = [
    {'pesel': '61041212345', 'firstname': 'José', 'lastname': 'Jiménez'},
    {'pesel': '61041212346', 'firstname': 'Jan', 'lastname': 'Twardowski'},
    {'pesel': '61041212347', 'firstname': 'Melissa', 'lastname': 'Lewis'},
    {'pesel': '61041212348', 'firstname': 'Alex', 'lastname': 'Vogel'},
    {'pesel': '61041212349', 'firstname': 'Ryan', 'lastname': 'Stone'},
]


with sqlite3.connect(':memory:') as db:
    db.execute(SQL_CREATE_TABLE)
    db.executemany(SQL_INSERT, astronauts)
    cursor = db.cursor()

    for row in cursor.execute(SQL_SELECT):
        print(row)

# (1, 61041212345, 'José', 'Jiménez')
# (2, 61041212346, 'Jan', 'Twardowski')
# (3, 61041212347, 'Melissa', 'Lewis')
# (4, 61041212348, 'Alex', 'Vogel')
# (5, 61041212349, 'Ryan', 'Stone')

2.7. Use cases

Listing 2.38. Zapisywanie do bazy danych wyników pobranych z sensorów podłączonych po USB
#!/usr/bin/env python3

import json
import datetime
import logging
import sqlite3
import serial


logging.basicConfig(
    format='[%(asctime).19s] %(levelname)s %(message)s',
    level=logging.INFO
)


DATABASE = '/home/pi/database/sensor-data.sqlite3'
DEVICE = '/dev/ttyACM0'
MEASUREMENTS = {
    'air_temperature': 'C',
    'air_humidity': '%',
    'water_temperature': 'C',
    'luminosity': 'lux',
    'power_k1': 'on/off',
    'power_k2': 'on/off',
    'power_k3': 'on/off',
    'power_k4': 'on/off',
}


with sqlite3.connect(DATABASE) as db:
    db.execute("""CREATE TABLE IF NOT EXISTS sensor_data (
        datetime DATETIME PRIMARY KEY,
        sync_datetime DATETIME DEFAULT NULL,
        device VARCHAR(255),
        parameter VARCHAR(255),
        value REAL,
        unit VARCHAR(255));""")
    db.execute('CREATE UNIQUE INDEX IF NOT EXISTS sensor_data_datetime_index ON sensor_data (datetime);')
    db.execute('CREATE INDEX IF NOT EXISTS sensor_data_sync_datetime_index ON sensor_data (sync_datetime);')


def save_to_sqlite3(data):
    for parameter, value in data.items():
        unit = MEASUREMENTS.get(parameter, None)

        with sqlite3.connect(DATABASE) as db:
            db.execute('INSERT INTO sensor_data VALUES (:datetime, NULL, :device, :parameter, :value, :unit)', {
                'datetime': datetime.datetime.now(datetime.timezone.utc),
                'parameter': parameter,
                'value': float(value),
                'unit': unit,
                'device': 'hydroponics',
            })


if __name__ == '__main__':
    with serial.Serial(port=DEVICE, baudrate=115200) as ser:
        while True:
            line = ser.readline()
            try:
                data = json.loads(line)
                save_to_sqlite3(data)
                logging.info(data)
            except json.decoder.JSONDecodeError:
                logging.error(line)

2.8. Assignments

2.8.1. Iris Database

  • Filename: database_iris.py

  • Lines of code to write: 30 lines

  • Estimated time of completion: 30 min

  • Input data: Listing 2.39.

Listing 2.39. Iris Database
4.3,3.0,1.1,0.1,0
5.8,4.0,1.2,0.2,0
5.7,4.4,1.5,0.4,1
5.4,3.9,1.3,0.4,2
5.1,3.5,1.4,0.3,1
5.7,3.8,1.7,0.3,0
5.1,3.8,1.5,0.3,0
5.4,3.4,1.7,0.2,1
5.1,3.7,1.5,0.4,0
4.6,3.6,1.0,0.2,0
5.1,3.3,1.7,0.5,2
4.8,3.4,1.9,0.2,0
5.0,3.0,1.6,0.2,1
5.0,3.4,1.6,0.4,2
5.2,3.5,1.5,0.2,1
5.2,3.4,1.4,0.2,2
4.7,3.2,1.6,0.2,0
  1. Skopiuj dane z listingu Listing 2.39. i zapisz je w pliku iris-db.csv

  2. Wczytaj dane z pliku iris-db.csv

  3. Stwórz species: Dict[int, str] o strukturze:

    • 0 - setosa

    • 1 - versicolor

    • 2 - virginica

  4. Bazę pomiarów Irysów przekonwertuj na tabelę w sqlite3

  5. Do połączenia wykorzystaj context manager (with)

  6. Dane wrzuć do bazy za pomocą .executemany() podając dict

  7. Nazwy poszczególnych kolumn:

    • id - int

    • species - str - podmień int na str ze słownika species

    • datetime - datetime - z datą i czasem dodania w UTC

    • sepal_length - float

    • sepal_width - float

    • petal_length - float

    • petal_width - float

  8. Przy dodawaniu danych species podmień int na str ze słownika species

  9. Dodaj kolumnę datetime z datą i czasem dodania (UTC)

  10. Załóż index na datetime

  11. Wyniki wypisz z bazy danych (SELECT * FROM iris ORDER BY datetime DESC)

  12. Zwracaj dane jako sqlite3.Row

The whys and wherefores
  • Parsowanie plików csv

  • Wykorzystywanie bazy sqlite3

  • Tworzenie bazy danych

  • Zakładanie indeksów na bazie danych

  • Wrzucanie danych do bazy

  • Wyciąganie danych z bazy

  • Konwersja typów

  • Korzystanie z datetime

2.8.2. Tworzenie bazy danych i proste zapytania

  • Filename: database_addressbook.py

  • Lines of code to write: 15 lines

  • Estimated time of completion: 20 min

  1. Wykorzystaj kod z listingu Listing 2.40. oraz Listing 2.41.

  2. Wykorzystaj cursor oraz połączenia jako context manager (with)

  3. Dane powinny być zwracane w postaci listy dict

  4. Do wpisywania danych wykorzystaj konstrukcję execute wykorzystując dict jako argument

Zadanie z gwiazdką
  • Dodaj obsługę wielu adresów

  • Dodaj obsługę relacji w aplikacji

Listing 2.40. Address Book SQL queries
CREATE TABLE IF NOT EXISTS contact (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    created DATETIME,
    modified DATETIME,
    first_name TEXT,
    last_name TEXT,
    date_of_birth DATE
);

CREATE UNIQUE INDEX IF NOT EXISTS last_name_index ON contact (last_name);
CREATE INDEX IF NOT EXISTS modified_index ON contact (modified);

CREATE TABLE IF NOT EXISTS address (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    contact_id INTEGER,
    street TEXT,
    city TEXT,
    state TEXT,
    code INT,
    country TEXT
);

INSERT INTO contact VALUES (
    NULL,
    :created,
    :modified,
    :first_name,
    :last_name,
    :date_of_birth
);

INSERT INTO address VALUES (
    NULL,
    :contact_id
    :street,
    :city,
    :state,
    :code,
    :country
);

UPDATE contact SET
    first_name=:firstname,
    last_name=:lastname,
    modified=:modified
WHERE id=:id;

SELECT * FROM contact;
Listing 2.41. Address Book data
José, Jiménez
    2101 E NASA Pkwy, 77058, Houston, Texas, USA
    , Kennedy Space Center, 32899, Florida, USA

Mark, Watney
    4800 Oak Grove Dr, 91109, Pasadena, California, USA
    2825 E Ave P, 93550, Palmdale, California, USA

Иван, Иванович
    Kosmodrom Bajkonur, Bajkonur, Kazachstan

Melissa Lewis,
    <NO ADDRESS>

Alex Vogel
    Linder Hoehe, 51147, Köln, Germany

2.8.3. Dane w jednej tabeli

  • Lines of code to write: 0 lines

  • Estimated time of completion: 5 min

  • Input data: Listing 2.41.

  1. Jak zapisać w jednej tabeli bez wykorzystania relacji?

  2. Przeprowadź dyskusję na temat zalet i wad podejść:

    • jedno pole adresy i dane rozdzielone separatorem (;)

    • wiele adresów i dane rozdzielone separatorem (;)

    • wiele kolumn dla każdego pola

  3. Które podejście jest najlepsze?

  4. Które podejście jest zgodne z ACID?