5.1. SimpleSQLite class

class simplesqlite.SimpleSQLite(database_src: Connection | SimpleSQLite | str, mode: str = 'a', delayed_connection: bool = True, max_workers: int | None = None, profile: bool = False, **connect_kwargs: Any)[source]

Wrapper class for sqlite3 module.

Parameters:
  • database_src (str) – SQLite database source. Acceptable types are: (1) File path to a database to be connected. (2) sqlite3.Connection instance. (3) SimpleSQLite instance

  • mode (str) – Open mode.

  • delayed_connection (bool) – Delaying connection to a database until access to the database the first time, if the value is True.

  • max_workers (int) – Maximum number of workers to generate a table. In default, the same as the total number of CPUs.

  • profile (bool) – Recording SQL query execution time profile, if the value is True.

  • **connect_kwargs (Any) –

    Keyword arguments passing to sqlite3.connect.

check_connection() None[source]
Raises:

simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.

Sample Code:
import simplesqlite

con = simplesqlite.SimpleSQLite("sample.sqlite", "w")

print("---- connected to a database ----")
con.check_connection()

print("---- disconnected from a database ----")
con.close()
try:
    con.check_connection()
except simplesqlite.NullDatabaseConnectionError as e:
    print(e)
Output:
---- connected to a database ----
---- disconnected from a database ----
null database connection
close() None[source]

Commit and close the connection.

commit() None[source]
connect(database_path: str, mode: str = 'a') None[source]

Connect to a SQLite database.

Parameters to open a connection to an SQLite database are passed via connect_kwargs argument of the constructor.

Parameters:
  • database_path (str) – Path to the SQLite database file to be connected.

  • mode (str) – "r": Open for read only. "w": Open for read/write. Delete existing tables when connecting. "a": Open for read/write. Append to the existing tables.

Raises:
property connection: Connection | None

sqlite3.Connection instance of the connected database. :rtype: sqlite3.Connection

Type:

return

create_index(table_name: str, attr_name: str) None[source]
Parameters:
  • table_name (str) – Table name that contains the attribute to be indexed.

  • attr_name (str) – Attribute name to create index.

Raises:
create_index_list(table_name: str, attr_names: Sequence[str]) None[source]
Parameters:
  • table_name (str) – Table name that exists attribute.

  • attr_names (list) – List of attribute names to create indices. Ignore attributes that are not existing in the table.

See also

create_index()

create_table(table_name: str, attr_descriptions: Sequence[str]) bool[source]
Parameters:
  • table_name (str) – Table name to create.

  • attr_descriptions (list) – List of table description.

Raises:
create_table_from_csv(csv_source: str, table_name: str = '', attr_names: Sequence[str] = (), delimiter: str = ',', quotechar: str = '"', encoding: str = 'utf-8', primary_key: str | None = None, add_primary_key_column: bool = False, index_attrs: Sequence[str] | None = None) None[source]

Create a table from a CSV file/text.

Parameters:
  • csv_source (str) – Path to the CSV file or CSV text.

  • table_name (str) – Table name to create. Using CSV file basename as the table name if the value is empty.

  • attr_names (list) – Attribute names of the table. Use the first line of the CSV file as attributes if attr_names is empty.

  • delimiter (str) – A one-character string used to separate fields.

  • quotechar (str) – A one-character string used to quote fields containing special characters, such as the delimiter or quotechar, or which contain new-line characters.

  • encoding (str) – CSV file encoding.

  • primary_key (str) – Primary key of the creating table.

  • index_attrs (tuple) – List of attribute names that creating indices.

Raises:

ValueError – If the CSV data is invalid.

Dependency Packages:
Example:

Create a table from CSV

See also

create_table_from_data_matrix() csv.reader() pytablereader.CsvTableFileLoader.load() pytablereader.CsvTableTextLoader.load()

create_table_from_data_matrix(table_name: str, attr_names: Sequence[str], data_matrix: Any, primary_key: str | None = None, add_primary_key_column: bool = False, index_attrs: Sequence[str] | None = None, type_hints: Sequence[Type[AbstractType] | None] | None = None) None[source]

Create a table if not exists. Moreover, insert data into the created table.

Parameters:
  • table_name (str) – Table name to create.

  • attr_names (list) – Attribute names of the table.

  • data_matrix (List of dict/namedtuple()/list/tuple) – Data to be inserted into the table.

  • primary_key (str) – Primary key of the creating table.

  • index_attrs (tuple) – List of attribute names that creating indices.

Raises:
  • simplesqlite.NameValidationError – If the name is invalid for a SQLite table name.

  • simplesqlite.NameValidationError – If the name is invalid for a SQLite attribute name.

  • ValueError – If the data_matrix is empty.

Example:

Create a table from a data matrix

create_table_from_dataframe(dataframe: pandas.DataFrame, table_name: str = '', primary_key: str | None = None, add_primary_key_column: bool = False, index_attrs: Sequence[str] | None = None) None[source]

Create a table from a pandas.DataFrame instance.

Parameters:
  • dataframe (pandas.DataFrame) – DataFrame instance to convert.

  • table_name (str) – Table name to create.

  • primary_key (str) – Primary key of the creating table.

  • index_attrs (tuple) – List of attribute names that creating indices.

Examples:

Create a table from pandas DataFrame

create_table_from_json(json_source: str, table_name: str = '', primary_key: str | None = None, add_primary_key_column: bool = False, index_attrs: Sequence[str] | None = None) None[source]

Create a table from a JSON file/text.

Parameters:
  • json_source (str) – Path to the JSON file or JSON text.

  • table_name (str) – Table name to create.

  • primary_key (str) – Primary key of the creating table.

  • index_attrs (tuple) – List of attribute names that creating indices.

Dependency Packages:
Examples:

Create table(s) from JSON

See also

pytablereader.JsonTableFileLoader.load() pytablereader.JsonTableTextLoader.load()

create_table_from_tabledata(table_data: TableData, primary_key: str | None = None, add_primary_key_column: bool = False, index_attrs: Sequence[str] | None = None) None[source]

Create a table from tabledata.TableData.

Parameters:
  • table_data (tabledata.TableData) – Table data to create.

  • primary_key (str) – Primary key of the creating table.

  • index_attrs (tuple) – List of attribute names that creating indices.

property database_path: str | None

File path of the connected database. :rtype: str

Examples:
>>> from simplesqlite import SimpleSQLite
>>> con = SimpleSQLite("sample.sqlite", "w")
>>> con.database_path
'/tmp/sample.sqlite'
>>> con.close()
>>> print(con.database_path)
None
Type:

return

delete(table_name: str, where: str | Where | And | Or | None = None) Cursor | None[source]

Send a DELETE query to the database.

Parameters:
  • table_name (str) – Table name of executing the query.

  • where (str/Where/And/Or) – WHERE clause for the query.

drop_table(table_name: str) None[source]
Parameters:

table_name (str) – Table name to drop.

Raises:
execute_query(query: str | QueryItem, caller: Tuple | None = None) Cursor | None[source]

Send arbitrary SQLite query to the database.

Parameters:
Returns:

The result of the query execution.

Return type:

sqlite3.Cursor

Raises:

Warning

This method can execute an arbitrary query. i.e. No access permissions check by mode.

fetch_attr_names(table_name: str) List[str][source]
Returns:

List of attribute names in the table.

Return type:

list

Raises:
Example:
import simplesqlite

table_name = "sample_table"
con = simplesqlite.SimpleSQLite("sample.sqlite", "w")
con.create_table_from_data_matrix(
    table_name,
    ["attr_a", "attr_b"],
    [[1, "a"], [2, "b"]])

print(con.fetch_attr_names(table_name))

try:
    print(con.fetch_attr_names("not_existing"))
except simplesqlite.TableNotFoundError as e:
    print(e)
Output:
['attr_a', 'attr_b']
'not_existing' table not found in /tmp/sample.sqlite
fetch_attr_type(table_name: str) Dict[str, str][source]
Returns:

Dictionary of attribute names and attribute types in the table.

Return type:

dict

Raises:
fetch_num_records(table_name: str, where: str | Where | And | Or | None = None) int | None[source]

Fetch the number of records in a table.

Parameters:
  • table_name (str) – Table name to get number of records.

  • where (str/Where/And/Or) – WHERE clause for the query.

Returns:

Number of records in the table. None if no value matches the conditions, or the table not found in the database.

Return type:

int

fetch_sqlite_master() List[Dict][source]

Get sqlite_master table information as a list of dictionaries.

Returns:

sqlite_master table information.

Return type:

list

Raises:

simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.

Sample Code:
import json

from simplesqlite import SimpleSQLite

con = SimpleSQLite("sample.sqlite", "w")
data_matrix = [
    [1, 1.1, "aaa", 1,   1],
    [2, 2.2, "bbb", 2.2, 2.2],
    [3, 3.3, "ccc", 3,   "ccc"],
]
con.create_table_from_data_matrix(
    "sample_table",
    ["a", "b", "c", "d", "e"],
    data_matrix,
    index_attrs=["a"])

print(json.dumps(con.fetch_sqlite_master(), indent=4))
Output:
[
    {
        "tbl_name": "sample_table",
        "sql": "CREATE TABLE 'sample_table' ('a' INTEGER, 'b' REAL, 'c' TEXT, 'd' REAL, 'e' TEXT)",
        "type": "table",
        "name": "sample_table",
        "rootpage": 2
    },
    {
        "tbl_name": "sample_table",
        "sql": "CREATE INDEX sample_table_a_index ON sample_table('a')",
        "type": "index",
        "name": "sample_table_a_index",
        "rootpage": 3
    }
]
fetch_table_names(include_system_table: bool = False, include_view: bool = True) List[str][source]
Returns:

List of table names in the database.

Return type:

list

Raises:
Sample Code:
from simplesqlite import SimpleSQLite

con = SimpleSQLite("sample.sqlite", "w")
con.create_table_from_data_matrix(
    "hoge",
    ["attr_a", "attr_b"],
    [[1, "a"], [2, "b"]])
print(con.fetch_table_names())
Output:
['hoge']
fetch_value(select: str, table_name: str, where: str | Where | And | Or | None = None, extra: str | None = None) int | None[source]

Fetch a value from the table. Return None if no value matches the conditions, or the table not found in the database.

Parameters:
  • select (str) – Attribute for SELECT query

  • table_name (str) – Table name of executing the query.

  • where (str/Where/And/Or) – WHERE clause for the query.

Returns:

Result of execution of the query.

Raises:
fetch_view_names() List[str][source]
Returns:

List of table names in the database.

Return type:

list

get_profile(profile_count: int = 50) List[Any][source]

Get profile of query execution time.

Parameters:

profile_count (int) – Number of profiles to retrieve, counted from the top query in descending order by the cumulative execution time.

Returns:

Profile information for each query.

Return type:

list of namedtuple()

Raises:
Example:

Profiling

has_attr(table_name: str, attr_name: str | None) bool[source]
Parameters:
  • table_name (str) – Table name that the attribute exists.

  • attr_name (str) – Attribute name to be tested.

Returns:

True if the table has the attribute.

Return type:

bool

Raises:

simplesqlite.TableNotFoundError – If the table not found in the database.

Sample Code:
import simplesqlite

table_name = "sample_table"
con = simplesqlite.SimpleSQLite("sample.sqlite", "w")
con.create_table_from_data_matrix(
    table_name,
    ["attr_a", "attr_b"],
    [[1, "a"], [2, "b"]])

print(con.has_attr(table_name, "attr_a"))
print(con.has_attr(table_name, "not_existing"))
try:
    print(con.has_attr("not_existing_table", "attr_a"))
except simplesqlite.DatabaseError as e:
    print(e)
Output:
True
False
'not_existing' table not found in /tmp/sample.sqlite
has_attrs(table_name: str, attr_names: Sequence[str]) bool[source]
Parameters:
  • table_name (str) – Table name that attributes exists.

  • attr_names – Attribute names to tested.

Returns:

True if the table has all of the attribute.

Return type:

bool

Raises:

simplesqlite.TableNotFoundError – If the table not found in the database.

Sample Code:
import simplesqlite

table_name = "sample_table"
con = simplesqlite.SimpleSQLite("sample.sqlite", "w")
con.create_table_from_data_matrix(
    table_name,
    ["attr_a", "attr_b"],
    [[1, "a"], [2, "b"]])

print(con.has_attrs(table_name, ["attr_a"]))
print(con.has_attrs(table_name, ["attr_a", "attr_b"]))
print(con.has_attrs(table_name, ["attr_a", "attr_b", "not_existing"]))
try:
    print(con.has_attrs("not_existing_table", ["attr_a"]))
except simplesqlite.DatabaseError as e:
    print(e)
Output:
True
True
False
'not_existing' table not found in /tmp/sample.sqlite
has_table(table_name: str, include_view: bool = True) bool[source]
Parameters:

table_name (str) – Table name to be tested.

Returns:

True if the database has the table.

Return type:

bool

Sample Code:
from simplesqlite import SimpleSQLite

con = SimpleSQLite("sample.sqlite", "w")
con.create_table_from_data_matrix(
    "hoge",
    ["attr_a", "attr_b"],
    [[1, "a"], [2, "b"]])

print(con.has_table("hoge"))
print(con.has_table("not_existing"))
Output:
True
False
has_view(view_name: str) bool[source]
Parameters:

table_name (str) – Table name to be tested.

Returns:

True if the database has the table.

Return type:

bool

insert(table_name: str, record: Any, attr_names: Sequence[str] | None = None) bool[source]

Send an INSERT query to the database.

Parameters:
Raises:
Example:

Insert records into a table

insert_many(table_name: str, records: Sequence[Dict | Sequence], attr_names: Sequence[str] | None = None) int[source]

Send an INSERT query with multiple records to the database.

Parameters:
Returns:

Number of inserted records.

Return type:

int

Raises:
Example:

Insert records into a table

is_connected() bool[source]
Returns:

True if the connection to a database is valid.

Return type:

bool

Examples:
>>> from simplesqlite import SimpleSQLite
>>> con = SimpleSQLite("sample.sqlite", "w")
>>> con.is_connected()
True
>>> con.close()
>>> con.is_connected()
False
property mode: str | None

Connection mode: "r"/"w"/"a". :rtype: str

See also

connect()

Type:

return

rollback() None[source]
select(select: str | AttrList, table_name: str, where: str | Where | And | Or | None = None, extra: str | None = None) Cursor | None[source]

Send a SELECT query to the database.

Parameters:
  • select – Attribute for the SELECT query.

  • table_name (str) – Table name of executing the query.

  • where (str/Where/And/Or) – WHERE clause for the query.

  • extra (str) – Any other SQL clause for the query.

Returns:

Result of the query execution.

Return type:

sqlite3.Cursor

Raises:
select_as_dataframe(table_name: str, columns: Sequence[str] | None = None, where: str | Where | And | Or | None = None, extra: str | None = None) pandas.DataFrame[source]

Get data in the database and return fetched data as a pandas.Dataframe instance.

Parameters:
  • table_name (str) – Table name of executing the query.

  • columns – Column names to get data. If the value is None, get data from all of the columns in the table.

  • whereWHERE clause for the query.

  • extra – Any other SQL clause for the query.

Returns:

Table data as a pandas.Dataframe instance.

Return type:

pandas.DataFrame

Raises:
Example:

Get Data from a table as pandas DataFrame

Note

pandas package required to execute this method.

select_as_dict(table_name: str, columns: Sequence[str] | None = None, where: str | Where | And | Or | None = None, extra: str | None = None) List[OrderedDict[str, Any]] | None[source]

Get data in the database and return fetched data as a collections.OrderedDict list.

Parameters:
  • table_name (str) – Table name of executing the query.

  • columns (list) – Column names to get data. If the value is None, get data from all of the columns in the table.

  • where (str/Where/And/Or) – WHERE clause for the query.

  • extra (str) – Any other SQL clause for the query.

Returns:

Table data as collections.OrderedDict instances.

Return type:

list of collections.OrderedDict

Raises:
Example:

Get Data from a table as OrderedDict

select_as_memdb(table_name: str, columns: Sequence[str] | None = None, where: str | Where | And | Or | None = None, extra: str | None = None) SimpleSQLite[source]

Get data in the database and return fetched data as a in-memory SimpleSQLite instance.

Parameters:
  • table_name (str) – Table name of executing the query.

  • columns – Column names to get data. If the value is None, get data from all of the columns in the table.

  • where (str/Where/And/Or) – WHERE clause for the query.

  • extra (str) – Any other SQL clause for the query.

Returns:

Table data as a SimpleSQLite instance that connected to in memory database.

Return type:

SimpleSQLite

Raises:
select_as_tabledata(table_name: str, columns: Sequence[str] | None = None, where: str | Where | And | Or | None = None, extra: str | None = None, type_hints: Dict[str, Type[AbstractType] | None] | None = None) TableData[source]

Get data in the database and return fetched data as a tabledata.TableData instance.

Parameters:
  • table_name (str) – Table name of executing the query.

  • columns – Column names to get data. If the value is None, get data from all of the columns in the table.

  • where (str/Where/And/Or) – WHERE clause for the query.

  • extra (str) – Any other SQL clause for the query.

Returns:

Table data as a tabledata.TableData instance.

Return type:

tabledata.TableData

Raises:

Note

pandas package required to execute this method.

set_row_factory(row_factory: Callable | None) None[source]

Set row_factory to the database connection.

property total_changes: int
update(table_name: str, set_query: str | Sequence[Set], where: str | Where | And | Or | None = None) Cursor | None[source]

Execute an UPDATE query.

Parameters:
  • table_name (str) – Table name of executing the query.

  • set_query (Union[str, Sequence[Set]]) – SET clause for the update query.

  • where (str/Where/And/Or , optional) – WHERE clause for the update query. Defaults to None.

Raises:
validate_access_permission(valid_permissions: Sequence[str]) None[source]
Parameters:

valid_permissions (list/tuple) – List of permissions that access is allowed.

Raises:
verify_attr_existence(table_name: str, attr_name: str) None[source]
Parameters:
  • table_name (str) – Table name that the attribute exists.

  • attr_name (str) – Attribute name to tested.

Raises:
Sample Code:
from simplesqlite import (
    SimpleSQLite,
    DatabaseError,
    AttributeNotFoundError
)

table_name = "sample_table"
con = SimpleSQLite("sample.sqlite", "w")
con.create_table_from_data_matrix(
    table_name,
    ["attr_a", "attr_b"],
    [[1, "a"], [2, "b"]])

con.verify_attr_existence(table_name, "attr_a")
try:
    con.verify_attr_existence(table_name, "not_existing")
except AttributeNotFoundError as e:
    print(e)
try:
    con.verify_attr_existence("not_existing", "attr_a")
except DatabaseError as e:
    print(e)
Output:
'not_existing' attribute not found in 'sample_table' table
'not_existing' table not found in /tmp/sample.sqlite
verify_table_existence(table_name: str, allow_view: bool = True) None[source]
Parameters:

table_name (str) – Table name to be tested.

Raises:
  • simplesqlite.TableNotFoundError – If the table not found in the database.

  • simplesqlite.NameValidationError – If the name is invalid for a SQLite table name.

Sample Code:
import simplesqlite

table_name = "sample_table"
con = simplesqlite.SimpleSQLite("sample.sqlite", "w")
con.create_table_from_data_matrix(
    table_name,
    ["attr_a", "attr_b"],
    [[1, "a"], [2, "b"]])

con.verify_table_existence(table_name)
try:
    con.verify_table_existence("not_existing")
except simplesqlite.DatabaseError as e:
    print(e)
Output:
'not_existing' table not found in /tmp/sample.sqlite