korone.database package

Submodules

class korone.database.connection.Connection(*args, **kwargs)

Bases: Protocol

Database connection.

Connection Classes should receive the DBMS-specific parameters directly through the __init__ method.

Examples

>>> class SQLite3Connection:
...     # SQLite3-specific parameters are
...     # passed through __init__
...     def __init__(self, path: str):
...         self.path = path
...
...     # Context Manager
...     async def __aenter__(self):
...         ...
...         self.connect()
...
...     async def __aexit__(self):
...         ...
...         self.close()
...
...     async def connect(self): ...
...     async def table(self, name: str) -> Table: ...
...     async def close(self): ...

async commit()

Commit the current transaction.

This method is used to commit the current transaction. If there is no current transaction, this method does nothing.

Raises:

RuntimeError – If the connection is not yet open.

async connect()

Open a connection to a database.

This method opens a connection to a database.

async execute(sql, parameters=(), /)

Execute SQL operations.

This method executes an SQL operation.

Parameters:

• sql (str) – The SQL statement to be executed.

• parameters (tuple, optional) – The parameters to be used in the SQL statement, by default ().

async table(name)

Return a Table, which can be used for database related operations.

This method returns a Table object, which can be used for database related operations.

Parameters:

name (str) – The name of the table.

Returns:

Table – A Table object representing the specified table.

async close()

Close the connection.

This method closes the connection to the database.

class korone.database.impl.SQLite3Table(*, conn, table)

Bases: object

Represent the specifics of a SQLitie3 Table.

This class is used internally by SQLite3Connection to perform operations on the database.

Parameters:

• conn (_Conn) – The connection object.

• table (str) – The name of the table.

async insert(fields)

Insert a row on the table.

This method inserts a row on the table.

Parameters:

fields (Document) – The fields to be inserted.

async query(query)

Query rows that match the criteria.

This method queries rows that match the criteria specified by the query.

Parameters:

query (Query) – The query that specifies the criteria.

Returns:

Documents – A list of Document objects representing the rows that match the criteria.

async update(fields, query)

Update fields on rows that match the criteria.

This method updates the fields of rows that match the criteria specified by the query.

Parameters:

• fields (Document) – The fields to be updated.

• query (Query) – The query that specifies the criteria.

async delete(query)

Delete rows that match the criteria.

This method deletes rows that match the criteria specified by the query.

Parameters:

query (Query) – The query that specifies the criteria.

class korone.database.impl.SQLite3Connection(*args, path='/home/docs/.local/share/korone/korone.sqlite', **kwargs)

Bases: object

Represent a connection to a SQLite database.

This class provides methods for connecting to a SQLite database, executing SQL statements, and managing the connection.

Parameters:

• *args – Additional positional arguments for the sqlite3.connect function.

• path (Path, optional) – The path to the SQLite database file, by default Path(DEFAULT_DBFILE_PATH).

• **kwargs – Additional keyword arguments for the sqlite3.connect function.

_path

The path to the SQLite database file.

Type:

Path

_args

Additional positional arguments for the sqlite3.connect function.

Type:

tuple

_kwargs

Additional keyword arguments for the sqlite3.connect function.

Type:

dict

_conn

The SQLite database connection object.

Type:

sqlite3.Connection | None

async commit()

Commit the current transaction.

This method is used to commit the current transaction. If there is no current transaction, this method does nothing.

Raises:

RuntimeError – If the connection is not yet open.

async connect()

Connect to the SQLite database.

This method connects to the SQLite database and stores the connection in the _conn attribute.

Raises:

RuntimeError – If the connection is already in place.

async table(name)

Return a Table object representing the specified table.

This methos returns a Table object representing the specified table. The Table object can be used to perform queries and other operations on the table.

Parameters:

name (str) – The name of the table.

Returns:

Table – A Table object representing the specified table.

async execute(sql, parameters=(), /, script=False)

Execute an SQL statement with optional parameters.

Executes an SQL statement with optional parameters.

Parameters:

• sql (str) – The SQL statement to be executed.

• parameters (tuple, optional) – The parameters to be used in the SQL statement, by default ().

• script (bool, optional) – If True, the SQL statement is treated as a script, by default False.

Raises:

RuntimeError – Raised if the connection is not yet open.

async close()

Close the database connection.

This method is automatically called when the context manager exits.

Raises:

RuntimeError – If the connection is not yet open.

exception korone.database.query.MalformedQueryError

Bases: Exception

An exception raised when a query is not properly formed.

This exception is raised when a query is not properly formed, such as when an invalid operator is used or when a key is not a string.

class korone.database.query.Query(*, lhs=None, operator=None, rhs=None)

Bases: object

A class that provides a high-level interface for building SQL queries.

This class allows you to specify what element or elements to fetch from the database. It provides a higher level interface to the database by using queries, thereby preventing the user from dealing with SQL Queries directly.

The Query class uses operator overloading to build SQL queries. For example, the == operator is overloaded by the __eq__ method to create a SQL equality condition. The & and | operators are overloaded to create SQL AND and OR conditions, respectively.

The __getattr__ and __getitem__ methods are used to specify the column names in the SQL query. They set the left-hand side of the query condition.

The __copy__ method is used to create a copy of a Query instance. This is useful when you want to create a new query that is similar to an existing one.

The _new_node method is used internally to create a new Query instance with the provided left-hand side, operator, and right-hand side. It is recommended to use this method instead of directly creating a Query instance to ensure a defined state.

The compile method is used to compile the Query instance into a SQL clause and its bound data. It returns a tuple containing the SQL clause and the bound data, which is used internally to execute the appropriate SQL statement.

Parameters:

• lhs (str, optional) – Left-hand side of the query.

• operator (str, optional) – Operator of the query.

• rhs (typing.Any, optional) – Right-hand side of the query.

Examples

>>> async with SQLite3Connection() as conn:
...     table = await conn.table("Users")
...     logician = Query()
...     await table.query(logician.name == "Kazimierz Kuratowski")
[{'desc': 'Polish mathematician and logician. [...]',
    'name': 'Kazimierz Kuratowski'}]

lhs

operator

rhs

compile()

Compile a Query to SQL Clause and its Bound Data.

This method will return a tuple containing the SQL Clause and the Bound Data, which is used internally to execute the appropriate SQL statement.

Returns:

CompiledQuery – A SQL Clause with Bound Data.

Raises:

MalformedQueryError – If the query is not properly formed.

Examples

>>> query = Query()
>>> query.name == "John"
>>> query.compile()
('name == ?', ('John',))

class korone.database.table.Document

Bases: dict[str, Any]

Document represents a single row on the SQL Database Table.

One should note that, due to the limitation of Table Rows, one cannot use a Collection or Mapping as a Document Value.

Examples

>>> invaliddoc: Document = {"key": [1, 2, 3, 4, 5, 6]}

class korone.database.table.Documents

A list of Documents.

alias of list[Document]

class korone.database.table.Table(*args, **kwargs)

Bases: Protocol

Table from the database.

It provides a higher level interface to the database by using queries, thereby preventing the user from dealing with SQL Queries directly.

async insert(fields)

Insert a row on the table.

The fields parameter may be of any type that contains object.__dict__. It may also be a Document type.

Keep in mind that, in the former case, keys starting with _ will be ignored, whereas in the latter, they will not.

Parameters:

fields (Document) – Fields to insert.

Notes

Document values cannot have nested values, due to the limitation of database rows. Refer to the Document type for more information.

Examples

>>> class HappyLittleCustomer:
...     def __init__(self, name: str):
...         self._name = name
>>> gummikunde = HappyLittleCustomer("nibble")
>>> vars(gummikunde)
{'_name': 'nibble'}
>>> # it won't insert anything, since we are passing
>>> # a class instead of a Document
>>> table.insert(gummikunde)

async query(query)

Query rows that match the criteria.

This method will return a list of Documents, where each Document represents a row on the table.

Parameters:

query (Query) – Matching criteria.

Returns:

Documents – List of Documents of rows that matched the criteria.

async update(fields, query)

Update fields on rows that match the criteria.

The fields has a special behavior. You should check the method insert for more information.

Parameters:

• fields (Document) – Fields to update.

• query (Query) – Matching criteria.

async delete(query)

Delete rows that match the criteria.

This method will delete all rows that match the criteria.

Parameters:

query (Query) – Matching criteria.