LilDB 0.5.5

LilDB provides a simplified wrapper for SQLite3.

LilDB

LilDB provides a simplified wrapper for SQLite3.

Connection.

You can connect to the database in two ways: the usual way and using the context manager.

Usual way:

from lildb import DB

db = DB("local.db")

# Disconnect
db.close()

Context manager:

from lildb import DB


with DB("local.db") as db:
    # do anything
    ...
# Disconnect

DB automatically collects information about existing tables, and allows you to present data in the form of dict or dataclass.

By default db returns data as dict, you can change that with 'use_datacls' flag.

from lildb import DB

# Dict rows
db = DB("local.db")

# DataClass rows
db = DB("local.db", use_datacls=True)

About table

If you are not using a custom table (more on this below), then DB will collect data about the tables automatically and you can use them using the DB attributes. For example, if there is a 'Person' table in the database, then you can work with it through the 'person' attribute.

db = DB("local.db")

db.person
print(db.person)
# <Table: Person>

Create table

Simple create table without column types:

db.create_table("Person", ("name", "post", "email", "salary", "img"))

# Equivalent to 'CREATE TABLE IF NOT EXISTS Person(name, post, email, salary, img)'

Advanced create table

If you want use more features take this:

from lildb.column_types import Integer, Real, Text, Blob

db.create_table(
    "Person",
    {
        "id": Integer(primary_key=True),
        "name": Text(nullable=True),
        "email": Text(unique=True),
        "post": Text(default="Admin"),
        "salary": Real(default=10000),
        "img": Blob(nullable=True),
    },
)

# Equivalent to 'CREATE TABLE IF NOT EXISTS Person (id INTEGER PRIMARY KEY NOT NULL, name TEXT, email TEXT NOT NULL UNIQUE, post TEXT DEFAULT 'Admin' NOT NULL, salary REAL DEFAULT 10000 NOT NULL, img BLOB)'


db.create_table(
    "Post",
    {
        "id": Integer(),
        "name": Text(),
    },
    table_primary_key=("id", "name"),
)

# Equivalent to 'CREATE TABLE IF NOT EXISTS Post (id INTEGER NOT NULL, name TEXT NOT NULL, PRIMARY KEY(id,name))'

Insert data

Add new row:

db.person.insert({
    "name": "David",
    "email": "[e-mail]",
    "salary": 15.5,
    "post": "Manager",
})

# or
db.person.add({
    "name": "David",
    "email": "[e-mail]",
    "salary": 15.5,
})

# Equivalent to 'INSERT INTO Person (name, email, salary) VALUES(?, ?, ?)'

Add many rows:

persons = [
    {"name": "Ann", "email": "[e-mail]", "salary": 15, "post": "Manager"},
    {"name": "Jim", "email": "[e-mail]", "salary": 10, "post": "Security"},
    {"name": "Sam", "email": "[e-mail]", "salary": 1.5, "post": "DevOps"},
]

db.person.insert(persons)

# or
db.person.add(persons)

Query

You can use 'query' to create a more complex sql-query. But it unstable.

# Fetch person table obj
person_tb = db.person

# 'all' and 'first' executing query

# Return all data from person table
db.person.query().all()

# Return first row from person table
db.person.query().first()


# Use id and name column
db.person.query(person_tb.c.id, person_tb.c.name)

# Use sql func on column
db.person.query(person_tb.c.name.length())
db.person.query(person_tb.c.name.lower())
db.person.query(person_tb.c.id.max())
db.person.query(person_tb.c.id.is_(None))
db.person.query(person_tb.c.id.in_([1, 2]))

# Use other funcs
from lildb.sql import func

db.person.query(func.abs(person_tb.c.id))
db.person.query(func.distinct(person_tb.c.name))
db.person.query(func.lower(person_tb.c.name))
db.person.query().where(
    func.like(person_tb.c.name, "Dav%") | (person_tb.c.id == 3)
)

db.person.query(person_tb.c.name.upper().label("upper_name"))
# SELECT UPPER(`Person`.name) AS upper_name FROM Person

# Return data with id = 1
db.person.query().where(id=1)
# Alternative
db.person.query().where(person_tb.c.id == 1)
db.person.query().where(condition="id = 1")

db.person.query().where(id=1, name="David", filter_operator="AND")

# Various conditions
db.person.query().where(
    (person_tb.c.name == "David") | (person_tb.c.id == 2)
)

query = db.person.query()
# Limit data
query.limit(10).offset(2)

# Group by data
query.group_by(person_tb.c.salary)

# Order data
query.order_by(person_tb.c.id)

# Check exists
query.exists()

# Check row count
query.count()

Select data

Get all data from table:

db.person.all()

# Equivalent to 'SELECT * FROM Person'

Get first three rows:

db.person.select(size=3)

Iterate through the table:

for row in db.person:
    row

Simple filter:

db.person.select(salary=10, post="DevOps")

# Equivalent to 'SELECT * FROM Person WHERE salary = 10 AND post = "DevOps"'

db.person.select(id=1, post="DevOps", operator="OR")

# Equivalent to 'SELECT * FROM Person WHERE salary = 10 OR post = "DevOps"'

Get one row by id or position if id does not exist:

db.person[1]

# or
db.person.get(id=1)
db.person.get(name="Ann")

Select specific columns:

db.person.select(columns=["name", "id"])

# Equivalent to 'SELECT name, id FROM Person'

For more complex queries, use:

db.person.select(condition="salary < 15")
# Equivalent to 'SELECT * FROM Person WHERE salary < 15'


db.person.select(columns=["name"], condition="salary < 15 or name = 'Ann'")
# Equivalent to 'SELECT name FROM Person WHERE salary < 15 or name = 'Ann''

Update data

Change one row"

row = db.person[1]

# if use dict row
row["post"] = "Developer"
row.change()

# if use data class row
row.post = "Developer"
row.change()

Update column value in all rows

db.person.update({"salary": 100})

# Change David post
db.person.update({"post": "Admin"}, id=1)

Simple filter

db.person.update({"post": "Developer", "salary": 1}, id=1, name="David")

db.person.update(
    {"post": "Admin", "salary": 1},
    name="Ann",
    id=1,
    operator="or",
)
# Equivalent to 'UPDATE Person SET post = "Ann", salary = 1 WHERE name = 'Ann' or id = 1'

Delete data

Delete one row

row = db.person[1]
row.delete()

Simple filter delete

db.person.delete(id=1, name="David")

Delete all rows with salary = 1

db.person.delete(salary=1)

db.person.delete(salary=10, name="Sam", operator="OR")
# Equivalent to 'DELETE FROM Person WHERE salary = 10 OR name = "Sam"'

Multithreaded

You can use multithreaded using ThreadDB, example:

from lildb import ThreadDB
from concurrent.futures import ThreadPoolExecutor, wait


db = ThreadDB("local.db")

db.create_table(
    "Person",
    {
        "id": Integer(primary_key=True),
        "name": Text(nullable=True),
        "email": Text(unique=True),
        "post": Text(default="Admin"),
        "salary": Real(default=10000),
        "img": Blob(nullable=True),
    },
)

persons = [
    {"name": "Sam", "email": "[e-mail]", "salary": 1.5, "post": "DevOps"},
    {"name": "Ann", "email": "[e-mail]", "salary": 15, "post": "Manager"},
    {"name": "Jim", "email": "[e-mail]", "salary": 10, "post": "Security"},
    {"name": "David", "email": "[e-mail]", "salary": 16, "post": "Developer"},
]

with ThreadPoolExecutor(max_workers=10) as executor:
    futures = [executor.submit(db.person.add, person) for person in persons]
    wait(futures)

# for close connection
db.close()

How it work

Singleton

The Singleton pattern restricts the instantiation of a class to just one object. When you create an instance of the ThreadDB class, it checks if an instance already exists for the specified database. If one does, it returns the existing instance; otherwise, it creates a new instance for the specified database. This design pattern is particularly useful for managing database connections, as it provides a centralized point of access.

Thread Safety

To ensure multi-thread safety and prevent potential deadlocks, lildb utilizes an execution pipe. Whenever CRUD methods (Create, Read, Update, Delete) or custom SQL queries are called, the execution requests are sent to this pipe instead of directly accessing the database.

1. When a CRUD method or custom SQL query is invoked, lildb places the request in a queue that serves as the execution pipe.
2. In a separate execution thread, the requests are processed one by one from the execution pipe.
3. The separate thread reads the requests and executes them sequentially on the SQLite database.

Custom rows, tables, db

If you want to create a custom class of rows or tables, then you can do it as follows:

# We create custom row for table Post
from lildb.rows import dataclass_row
from lildb import Table
from lildb import DB
from lildb.column_types import Integer
from lildb.column_types import Text


@dataclass_row
class CustomPostRow:
    """Any custom data class row."""

    id: int
    name: str

    def title_post(self) -> str:
        """Any custom method."""
        return self.name.title()


class CustomPostTable(Table):
    """Any custom table class."""

    # Table name in DB
    table_name = "post"

    # Use custom data class row
    row_cls = CustomPostRow


class CustomDB(DB):
    """Custom DB."""

    post = CustomPostTable()


# Work with custom obj
db = CustomDB("post.db")

# Create table
db.create_table(
    "Post",
    {
        "id": Integer(),
        "name": Text(),
    },
    table_primary_key=("id", "name"),
)


print(db.post)
# <CustomPostTable: Post>

db.post.add({"id": 1, "name": "manager"})
db.post.add({"id": 2, "name": "developer"})

print(db.post.all())
# [CustomPostRow(id=1, name=manager), CustomPostRow(id=2, name=developer)]


row = db.post.get(id=1)
print(row.title_post())
# Manager

row.name = "admin"
row.change()

print(row.title_post())
# Admin

row.delete()

print(db.post.all())
# [CustomPostRow(id=2, name=developer)]

dataclass_row

dataclass_row (from lildb.rows import dataclass_row) works the same way as 'dataclass' (from dataclasses), the only difference is that 'dataclass_row' adds two arguments and a mixin to work correctly.

If you don't want to use 'dataclass_row' then make your row-class as follows:

from dataclasses import dataclass
from lildb import _RowDataClsMixin
from lildb import Table


@dataclass
class CustomPostRow(_RowDataClsMixin):
    """Any custom data class row."""

    id: int
    name: str

    # Required fields for row-cls
    table: Table
    changed_columns: set

    def title_post(self) -> str:
        """Any custom method."""
        return self.name.title()

Custom Dict row

If you want to use dict instead of dataclass, you can do it like this

from lildb import RowDict


class CustomPostRow(RowDict):
    """Any custom data class row."""

    def title_post(self) -> str:
        """Any custom method."""
        return self["name"].title()

Custom select, insert, delete and update

The corresponding class is responsible for each CRUD operation. You can create your own instances in the following way

# operation classes
from lildb import Select, Insert, Update, Delete

class CustomSelect(Select):
    """Custom select."""

    def get_manager(self) -> list:
        """Get all managers."""
        return self(name="manager")


class CustomPostTable(Table):
    """Any custom table class."""

    # Table name in DB
    name = "post"

    # Use custom data class row
    row_cls = CustomPostRow

    # Custom select
    select = CustomSelect

    # Custom other operation
    # insert = CustomInsert
    # update = CustomUpdate
    # delete = CustomDelete