4.2. Database SQLite

4.2.1. Installation

  1. Check if Sqlite3 is already installed:

    $ sqlite3 --version
    
  2. If not, then:

    • Download from https://sqlite.org/download.html

    • Add to $PATH directory

      • C:\Program Files\Git\cmd on Windows

      • /usr/local/bin/ on *nix

    • Check if Sqlite3 is installed (Point 1.)

4.2.2. SQL Syntax

Note

For SQL Syntax refer to Database SQL

4.2.2.1. Data Types

Table 4.1. 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

4.2.2.2. Constrains

Table 4.2. 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

4.2.3. Connection

Listing 4.19. Connection to in-memory database
import sqlite3

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

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

4.2.4. Execute

Listing 4.21. Execute
import sqlite3


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

data = {'pesel': '61041212345', 'first_name': 'José', 'last_name': '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')

4.2.5. Executemany

4.2.5.1. List[tuple]

Listing 4.22. Execute many
import sqlite3


SQL_CREATE_TABLE = """
    CREATE TABLE IF NOT EXISTS astronauts (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        pesel INTEGER UNIQUE,
        first_name TEXT,
        last_name 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')

4.2.5.2. List[dict]

Listing 4.23. Execute many
import sqlite3


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

SQL_INSERT_DICT = 'INSERT INTO astronauts VALUES (NULL, :pesel, :first_name, :last_name)'


list_of_dicts = [
    {'pesel': '61041212345', 'first_name': 'José', 'last_name': 'Jiménez'},
    {'pesel': '61041212346', 'first_name': 'Jan', 'last_name': 'Twardowski'},
    {'pesel': '61041212347', 'first_name': 'Melissa', 'last_name': 'Lewis'},
    {'pesel': '61041212348', 'first_name': 'Alex', 'last_name': 'Vogel'},
    {'pesel': '61041212349', 'first_name': 'Ryan', 'last_name': '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')

4.2.6. Results

4.2.6.1. Fetch as List[tuple]

Listing 4.24. Results
import sqlite3


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


astronauts = [
    {'pesel': '61041212345', 'first_name': 'José', 'last_name': 'Jiménez'},
    {'pesel': '61041212346', 'first_name': 'Jan', 'last_name': 'Twardowski'},
    {'pesel': '61041212347', 'first_name': 'Melissa', 'last_name': 'Lewis'},
    {'pesel': '61041212348', 'first_name': 'Alex', 'last_name': 'Vogel'},
    {'pesel': '61041212349', 'first_name': 'Ryan', 'last_name': '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')

4.2.6.2. Fetch as List[dict]

Listing 4.25. Results with dict
import sqlite3


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


astronauts = [
    {'pesel': '61041212345', 'first_name': 'José', 'last_name': 'Jiménez'},
    {'pesel': '61041212346', 'first_name': 'Jan', 'last_name': 'Twardowski'},
    {'pesel': '61041212347', 'first_name': 'Melissa', 'last_name': 'Lewis'},
    {'pesel': '61041212348', 'first_name': 'Alex', 'last_name': 'Vogel'},
    {'pesel': '61041212349', 'first_name': 'Ryan', 'last_name': '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, 'first_name': 'José', 'last_name': 'Jiménez'}
# {'id': 2, 'pesel': 61041212346, 'first_name': 'Jan', 'last_name': 'Twardowski'}
# {'id': 3, 'pesel': 61041212347, 'first_name': 'Melissa', 'last_name': 'Lewis'}
# {'id': 4, 'pesel': 61041212348, 'first_name': 'Alex', 'last_name': 'Vogel'}
# {'id': 5, 'pesel': 61041212349, 'first_name': 'Ryan', 'last_name': 'Stone'}

4.2.7. Cursor

Listing 4.26. Results with cursor
import sqlite3


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


astronauts = [
    {'pesel': '61041212345', 'first_name': 'José', 'last_name': 'Jiménez'},
    {'pesel': '61041212346', 'first_name': 'Jan', 'last_name': 'Twardowski'},
    {'pesel': '61041212347', 'first_name': 'Melissa', 'last_name': 'Lewis'},
    {'pesel': '61041212348', 'first_name': 'Alex', 'last_name': 'Vogel'},
    {'pesel': '61041212349', 'first_name': 'Ryan', 'last_name': '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')

4.2.8. Use cases

Listing 4.27. 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)

4.2.9. Assignments

4.2.9.1. Database SQLite Iris

English
  1. Use data from "Input" section (see below)

  2. Save input data as database-sqlite-iris.csv file

  3. Read data from file (don't use csv or pandas library)

  4. Connect to the sqlite3 using context manager (with)

  5. Create table iris, column names are specified in "Input" section (see below)

  6. Replace int to str according to SPECIES conversion table (see input data)

  7. Save data to database table

  8. Print results using SELECT * FROM iris ORDER BY datetime DESC

Polish
  1. Użyj danych z sekcji "Input" (patrz poniżej)

  2. Zapisz dane wejściowe do pliku database-sqlite-iris.csv

  3. Wczytaj dane z pliku (nie używaj biblioteki csv lub pandas)

  4. Połącz się do bazy danych sqlite3 używając context managera (with)

  5. Stwórz tabelę iris o kolumnach podanych w sekcji "Input" (patrz poniżej)

  6. Podmień int na str zgodnie z tabelą podstawień SPECIES (patrz dane wejściowe)

  7. Zapisz dane do tabeli w bazie danych

  8. Wypisz wyniki z bazy danych SELECT * FROM iris ORDER BY datetime DESC

Non functional requirements
  • Use context manager (with) for connection and for opening file

  • Return data as sqlite3.Row

  • Add data in dict format using .executemany()

  • Save date and time to database in UTC

Extra task
  • Create index on datetime column

  • Use cursor

The whys and wherefores
  • Parsing csv files

  • Using sqlite3 database

  • Creating database

  • Inserting data to database

  • Selecting data from database

  • Type casting

  • Using datetime and UTC timezone

  • Creating indexes (extra task)

Input
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
Listing 4.28. Input Species substitution dict
SPECIES = {
    0: 'setosa',
    1: 'versicolor',
    2: 'virginica',
}
CREATE TABLE IF NOT EXISTS iris (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    species TEXT,
    datetime DATETIME,
    sepal_length REAL,
    sepal_width REAL,
    petal_length REAL,
    petal_width REAL
);
INSERT INTO iris VALUES (
    NULL,
    :species,
    :datetime,
    :sepal_length,
    :sepal_width,
    :petal_length,
    :petal_width
);
SELECT * FROM iris ORDER BY datetime DESC

4.2.9.2. Database SQLite Relations

English
  1. Use data from "Input" section (see below)

  2. Create database for input data

  3. Add support for many addresses

  4. Insert data to database

  5. Select data from database using JOIN relations

Polish
  1. Użyj danych z sekcji "Input" (patrz poniżej)

  2. Stwórz bazę danych na podstawie danych wejściowych

  3. Dodaj obsługę dla wielu adresów

  4. Zapisz dane do bazy

  5. Wypisz dane z bazy wykorzystując relację JOIN

Input
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
Hint
Listing 4.29. Hint
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;

4.2.9.3. Database SQLite Logs

English
  1. Use data from "Input" section (see below)

  2. Save input data to file apollo11-timeline.log

  3. Extract datetime object, level name and message from each line

  4. Collect data to DATA: List[dict] (see below)

  5. Create database schema for logs

  6. Add all logs to database

  7. Select only WARNING logs between 1969-07-20 and 1969-07-21

  8. Order logs by datetime descending

  9. Print result: List[dict]

  10. Compare result with "Output" section (see below)

Polish
  1. Użyj danych z sekcji "Input" (patrz poniżej)

  2. Zapisz dane wejściowe do pliku apollo11-timeline.log

  3. Wyciągnij obiekt datetime, poziom logowania oraz wiadomość z każdej linii

  4. Zbierz dane do DATA: List[dict] (patrz sekcja input)

  5. Stwórz schemat bazy danych dla logów

  6. Dodaj wszystkie linie do bazy danych

  7. Wybierz tylko logi WARNING z przedziału 1969-07-20 i 1969-07-21

  8. Posortuj logi w kolejności datetime malejąco

  9. Wyświetl result: List[dict]

  10. Porównaj wyniki z sekcją "Output" (patrz poniżej)

Input
DATA = """
    1969-07-14T21:00:00 [INFO] Terminal countdown started
    1969-07-16T13:31:53 [WARNING] S-IC engine ignition (#5)
    1969-07-16T13:33:23 [DEBUG] Maximum dynamic pressure (735.17 lb/ft^2)
    1969-07-16T13:34:44 [WARNING] S-II ignition
    1969-07-16T13:35:17 [DEBUG] Launch escape tower jettisoned
    1969-07-16T13:39:40 [DEBUG] S-II center engine cutoff
    1969-07-16T16:22:13 [INFO] Translunar injection
    1969-07-16T16:56:03 [INFO] CSM docked with LM/S-IVB
    1969-07-16T17:21:50 [INFO] Lunar orbit insertion ignition
    1969-07-16T21:43:36 [INFO] Lunar orbit circularization ignition
    1969-07-20T17:44:00 [INFO] CSM/LM undocked
    1969-07-20T20:05:05 [WARNING] LM powered descent engine ignition
    1969-07-20T20:10:22 [ERROR] LM 1202 alarm
    1969-07-20T20:14:18 [ERROR] LM 1201 alarm
    1969-07-20T20:17:39 [WARNING] LM lunar landing
    1969-07-21T02:39:33 [DEBUG] EVA started (hatch open)
    1969-07-21T02:56:15 [WARNING] 1st step taken lunar surface (CDR) "That's one small step for [a] man... one giant leap for mankind"
    1969-07-21T03:05:58 [DEBUG] Contingency sample collection started (CDR)
    1969-07-21T03:15:16 [INFO] LMP on lunar surface
    1969-07-21T05:11:13 [DEBUG] EVA ended (hatch closed)
    1969-07-21T17:54:00 [WARNING] LM lunar liftoff ignition (LM APS)
    1969-07-21T21:35:00 [INFO] CSM/LM docked
    1969-07-22T04:55:42 [WARNING] Transearth injection ignition (SPS)
    1969-07-24T16:21:12 [INFO] CM/SM separation
    1969-07-24T16:35:05 [WARNING] Entry
    1969-07-24T16:50:35 [WARNING] Splashdown (went to apex-down)
    1969-07-24T17:29 [INFO] Crew egress
"""
Output
result: List[dict] = [

    {'date': datetime.datetime(1969, 7, 21, 17, 54, 00, tzinfo=datetime.timezone.utc),
     'level': 'WARNING',
     'message': 'LM lunar liftoff ignition (LM APS)'},

    {'date': datetime.datetime(1969, 7, 21, 2, 56, 15, tzinfo=datetime.timezone.utc),
     'level': 'WARNING',
     'message': '1st step taken lunar surface (CDR) "That\'s one small step for [a] man... one giant leap for mankind"'},

    {'date': datetime.datetime(1969, 7, 20, 20, 17, 39, tzinfo=datetime.timezone.utc),
     'level': 'WARNING',
     'message': 'LM lunar landing'},

...]