13. Bazy Danych

13.1. sqlite3

13.1.1. Typy danych w SQLite

  • NULL. The value is a NULL 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.

13.1.2. SQLite Auto Increment

Code Listing 13.1. Auto Increment
CREATE TABLE people (
    id integer primary key auto increment,
    first_name varchar(20),
    last_name varchar(20)
);

INSERT INTO people VALUES ("José", "Jiménez");
INSERT INTO people (first_name, last_name) VALUES ("Max", "Peck");

SELECT * FROM people;

1|José|Jiménez
2|Max|Peck

13.1.3. Połączenie

Code Listing 13.2. Connection
import sqlite3

connection = sqlite3.connect('example.db')
connection = sqlite3.connect(":memory:")


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

13.1.4. Context manager

Code Listing 13.3. Context Manager
import sqlite3

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

SQL_INSERT = """
    INSERT INTO kontakty VALUES (
        NULL,
        :pesel,
        :firstname,
        :lastname
    )
"""

SQL_SELECT = """
    SELECT * FROM kontakty
"""

dane = {'pesel': '61041212345', 'firstname': 'Jose', 'lastname': 'Jimenez'},


with sqlite3.connect(':memory:') as connection:
    connection.execute(SQL_CREATE_TABLE)
    connection.execute(SQL_INSERT, dane)

    for row in connection.execute(SQL_INSERT, dane):
        print(dict(row))

13.1.5. Cursor

Code Listing 13.4. Cursor
import sqlite3

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

SQL_INSERT = """
    INSERT INTO kontakty VALUES (
        NULL,
        :pesel,
        :firstname,
        :lastname
    )
"""

ksiazka_adresowa = [
    {'pesel': '61041212345', 'firstname': 'Jose', 'lastname': 'Jimenez'},
    {'pesel': '61041212345', 'firstname': 'Max', 'lastname': 'Peck'},
]


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

    try:
        with connection.cursor() as cursor:
            cursor.executemany(SQL_INSERT, ksiazka_adresowa)
    except sqlite3.IntegrityError:
        print('Pesel need to be UNIQUE')

13.1.6. Execute Many

Code Listing 13.5. Execute many
import sqlite3

conn = sqlite3.connect('example.db')

# Never do this -- insecure!
symbol = 'RHAT'
conn.execute("SELECT * FROM stocks WHERE symbol='%s'" % symbol)

# Do this instead
symbol = ['RHAT']
conn.execute('SELECT * FROM stocks WHERE symbol=?', symbol)
conn.fetchone()

# Larger example that inserts many records at a time
purchases = [
    ('2006-03-28', 'BUY', 'IBM', 1000, 45.00),
    ('2006-04-05', 'BUY', 'MSFT', 1000, 72.00),
    ('2006-04-06', 'SELL', 'IBM', 500, 53.00),
]

conn.executemany('INSERT INTO stocks VALUES (?,?,?,?,?)', purchases)

13.1.7. Wyniki

Code Listing 13.6. Results
import sqlite3

con = sqlite3.connect(":memory:")
con.row_factory = sqlite3.Row


for row in con.execute('SELECT * FROM stocks ORDER BY price'):
        print(row)

# ('2006-01-05', 'BUY', 'RHAT', 100, 35.14)
# ('2006-03-28', 'BUY', 'IBM', 1000, 45.0)
# ('2006-04-06', 'SELL', 'IBM', 500, 53.0)
# ('2006-04-05', 'BUY', 'MSFT', 1000, 72.0)



cur = con.cursor()
cur.execute("select 'John' as name, 42 as age")

for row in cur:
    assert row[0] == row["name"]
    assert row["name"] == row["nAmE"]
    assert row[1] == row["age"]
    assert row[1] == row["AgE"]

13.1.8. Typy i konwertery

Code Listing 13.7. Converters
import sqlite3
import datetime

con = sqlite3.connect(":memory:", detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES)
cur = con.cursor()
cur.execute("create table test(d date, ts timestamp)")

today = datetime.date.today()
now = datetime.datetime.now()

cur.execute("insert into test(d, ts) values (?, ?)", (today, now))
cur.execute("select d, ts from test")
row = cur.fetchone()
print(today, "=>", row[0], type(row[0]))
print(now, "=>", row[1], type(row[1]))

cur.execute(
    'select current_date as "d [date]", current_timestamp as "ts [timestamp]"')
row = cur.fetchone()
print("current_date", row[0], type(row[0]))
print("current_timestamp", row[1], type(row[1]))

13.2. Przykład praktyczny

Code Listing 13.8. 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)

13.3. pyMySQL

$ pip install PyMySQL
Code Listing 13.9. PyMySQL
CREATE TABLE `users` (
    `id` int(11) NOT NULL AUTO_INCREMENT,
    `email` varchar(255) COLLATE utf8_bin NOT NULL,
    `password` varchar(255) COLLATE utf8_bin NOT NULL,
    PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COLLATE=utf8_bin
AUTO_INCREMENT=1 ;
Code Listing 13.10. PyMySQL
import pymysql.cursors

# Connect to the database
connection = pymysql.connect(
    host='localhost',
    user='user',
    password='passwd',
    db='db',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

try:
    with connection.cursor() as cursor:
        # Create a new record
        sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)"
        cursor.execute(sql, ('webmaster@python.org', 'very-secret'))

    # connection is not autocommit by default. So you must commit to save
    # your changes.
    connection.commit()

    with connection.cursor() as cursor:
        # Read a single record
        sql = "SELECT `id`, `password` FROM `users` WHERE `email`=%s"
        cursor.execute(sql, ('webmaster@python.org',))
        result = cursor.fetchone()
        print(result)
finally:
    connection.close()

13.4. psycopg2

$ pip install psycopg2
Code Listing 13.11. Psycopg2
import psycopg2

# Connect to an existing database
conn = psycopg2.connect("dbname=test user=postgres")

# Open a cursor to perform database operations
cur = conn.cursor()

# Execute a command: this creates a new table
cur.execute(
    "CREATE TABLE test (id serial PRIMARY KEY, num integer, data varchar);")

# Pass data to fill a query placeholders and let Psycopg perform
# the correct conversion (no more SQL injections!)
cur.execute("INSERT INTO test (num, data) VALUES (%s, %s)",
                 ...(100, "abc'def"))

# Query the database and obtain data as Python objects
cur.execute("SELECT * FROM test;")
cur.fetchone()
# (1, 100, "abc'def")

# Make the changes to the database persistent
conn.commit()

# Close communication with the database
cur.close()
conn.close()

13.5. pymongo

$ python -m pip install pymongo
>>> from pymongo import MongoClient

>>> client = MongoClient('mongodb://localhost:27017/')
>>> db = client.test_database

>>> import datetime
>>> post = {"author": "Mike",
...         "text": "My first blog post!",
...         "tags": ["mongodb", "python", "pymongo"],
...         "date": datetime.datetime.utcnow()}

>>> posts = db.posts
>>> post_id = posts.insert_one(post).inserted_id
>>> post_id
ObjectId('...')
>>> for post in posts.find():
...   pprint.pprint(post)

>>> for post in posts.find({"author": "Mike"}):
...   pprint.pprint(post)

13.6. SQLAlchemy

13.7. Zadania kontrolne

13.7.1. Tworzenie bazy danych i proste zapytania

Skrypt z książką adresową z poprzednich zadań przepisz tak, aby wykorzystywał bazę danych do składowania informacji:

  • Nie wykorzystuj relacji, a dane adresowe zapisz zserializowane i rozdzielone średnikami ;
  • dla ułatwienia możesz przyjąć, że zawsze jest maks jeden adres
  • Wykorzystaj cursor
  • Dane powinny być zwracane dane w postaci listy dict
  • 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
Podpowiedź:
Code Listing 13.12. Ksiazka Adresowa zapytania SQL
SQL_CREATE_TABLE = """
    CREATE TABLE IF NOT EXISTS kontakty (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        firstname TEXT,
        lastname TEXT,
        adresy TEXT
    )
"""

SQL_INSERT = """
    INSERT INTO kontakty VALUES (
        NULL,
        :firstname,
        :lastname,
        :adresy
    )
"""

SQL_UPDATE = """
    UPDATE kontakty SET
        firstname=:firstname,
        lastname=:lastname,
        adresy=:adresy  
    WHERE id=:id
"""

SQL_SELECT = """
    SELECT * FROM kontakty
"""