Python SQLite3

Uma forma de trabalhar com SQLite3 usando a biblioteca padrão sqlite3.

DDL (Data Definition Language)

ddl.py

import sqlite3
from os import getenv


def create_table(sql: str):
    """Create a table in the database.

    :param sql: SQL code for table creation.
    """
    SQLITE_DATABASE_URI = getenv('SQLITE_DATABASE_URI', ':memory:')
    with sqlite3.connect(SQLITE_DATABASE_URI) as conector:
        cursor = conector.cursor()
        cursor.execute(sql)
        conector.commit()

DML - (Data Manipulation Language)

dml.py

import sqlite3
from functools import wraps
from os import getenv
from typing import Any, Callable, Optional


def execute(
    commit: Optional[bool] = True,
    fetchone: Optional[bool] = False,
    fetchmany: Optional[bool] = False,
    fetchall: Optional[bool] = False,
) -> Callable:
    """Executes a SQL statement.

    :param commit: Commit the current transaction. (default False)
    :param fetchone: fetches one row from the resultset. (default False)
    :param fetchmany: fetches all rows from the resultset. (default False)
    :param fetchall: fetches several rows from the resultset. (default False)
    """
    def wrap(func: Callable):
        @wraps(func)
        def wrapper(*args: list[Any], **kwargs: dict[Any, Any]):
            SQLITE_DATABASE_URI = getenv('SQLITE_DATABASE_URI', ':memory:')
            with sqlite3.connect(SQLITE_DATABASE_URI) as conector:
                cursor = conector.cursor()
                cursor.execute(func(*args, **kwargs))

                if fetchone:
                    result = cursor.fetchone()
                elif fetchall:
                    result = cursor.fetchall()
                elif fetchmany:
                    result = cursor.fetchmany()
                elif commit:
                    result = conector.commit()

                return result

        return wrapper
    return wrap


def _helper_sql(data: dict[int | str, Any]):
    parsed = ''
    for collumn, expression in data.items():
        parsed += f"{collumn} = '{expression}', "
    return parsed[:-2]


@execute(fetchone=True)
def select(table: str, where: dict[int | str, Any]):
    """Select data in a table.

    :param table: table name.
    :param where: filter records.
    """
    return f'SELECT * FROM {table} WHERE {_helper_sql(where)}'


@execute()
def insert(table: str, data: dict[int | str, Any]):
    """Insert data into the table.

    :param table: table name.
    :param data: the keys and values represent the INTO and VALUES clauses.

    """
    def parse_sql(data: dict[int | str, Any]) -> tuple[str, str]:
        into = ', '.join((str(key) for key in data.keys()))
        values = ', '.join((f"'{value}'" for value in data.values()))
        return into, values

    into, values = parse_sql(data)
    return f'INSERT INTO {table}({into}) VALUES({values})'


@execute()
def update(table: str, set: dict[int | str, Any], where: dict[int | str, Any]):
    """Update table data.

    :param table: table name.
    :param set: which columns and values are to be updated in a table.
    :param where: filter records.
    """
    return f'UPDATE {table} SET {_helper_sql(set)} WHERE {_helper_sql(where)}'


@execute()
def delete(table: str, where: dict[int | str, Any]):
    """Delete a user.

    :param table: table name.
    :param where: filter records.
    """
    return f"DELETE FROM {table} WHERE {_helper_sql(where)}"

Uso

main.py

import ddl
import dml

sql = """
CREATE TABLE users(id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
                   name TEXT NOT NULL,
                   phone TEXT NOT NULL,
                   email TEXT UNIQUE NOT NULL)
"""
ddl.create_table(sql)

dml.insert(
    'users',
    {
        'name': 'Igor1',
        'phone': '10987654321',
        'email': 'igor1@email.com',
    },
)
dml.select('users', {'id': 1})
dml.update('users', {'id': 1}, {'email': 'igor2@email.com'})
dml.delete('users', {'id': 1})

Todo

It would be interesting:

• Abstract sql from DDL
• Add operators other than = in where clauses

Comentários