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.
See also
- 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
- 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:
- Raises:
ValueError – If
database_path
is invalid ormode
is invalid.simplesqlite.DatabaseError – If the file is encrypted or is not a database.
simplesqlite.OperationalError – If unable to open the database file.
- 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:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.TableNotFoundError – If the table not found in the database.
- create_table(table_name: str, attr_descriptions: Sequence[str]) bool [source]¶
- Parameters:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
- 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
orquotechar
, 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:
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_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:
- Examples:
- 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:
- Dependency Packages:
- Examples:
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:
See also
- 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.
- drop_table(table_name: str) None [source]¶
- Parameters:
table_name (str) – Table name to drop.
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
- execute_query(query: str | QueryItem, caller: Tuple | None = None) Cursor | None [source]¶
Send arbitrary SQLite query to the database.
- Parameters:
query – Query to executed.
caller (tuple) – Caller information. Expects the return value of
logging.Logger.findCaller()
.
- Returns:
The result of the query execution.
- Return type:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.OperationalError – If failed to execute a query.
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:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.TableNotFoundError – If the table not found in the database.
simplesqlite.OperationalError – If failed to execute a query.
- 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:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.TableNotFoundError – If the table not found in the database.
simplesqlite.OperationalError – If failed to execute a query.
- fetch_num_records(table_name: str, where: str | Where | And | Or | None = None) int | None [source]¶
Fetch the number of records in a table.
- fetch_sqlite_master() List[Dict] [source]¶
Get sqlite_master table information as a list of dictionaries.
- Returns:
sqlite_master table information.
- Return type:
- 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:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.OperationalError – If failed to execute a query.
- 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:
- Returns:
Result of execution of the query.
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.OperationalError – If failed to execute a query.
- 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:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.OperationalError – If failed to execute a query.
- Example:
- has_attr(table_name: str, attr_name: str | None) bool [source]¶
- Parameters:
- Returns:
True
if the table has the attribute.- Return type:
- 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:
- 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:
- 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
- insert(table_name: str, record: Any, attr_names: Sequence[str] | None = None) bool [source]¶
Send an INSERT query to the database.
- Parameters:
table_name (str) – Table name of executing the query.
record (
dict
/namedtuple()
/list
/tuple
) – Record to be inserted.
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.OperationalError – If failed to execute a query.
- Example:
- 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:
table (str) – Table name of executing the query.
records (list of
dict
/namedtuple()
/list
/tuple
) – Records to be inserted.
- Returns:
Number of inserted records.
- Return type:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.TableNotFoundError – If the table not found in the database.
simplesqlite.OperationalError – If failed to execute a query.
- Example:
- 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:
- Returns:
Result of the query execution.
- Return type:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.TableNotFoundError – If the table not found in the database.
simplesqlite.OperationalError – If failed to execute a query.
- 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:
- Returns:
Table data as a
pandas.Dataframe
instance.- Return type:
pandas.DataFrame
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.TableNotFoundError – If the table not found in the database.
simplesqlite.OperationalError – If failed to execute a query.
- Example:
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:
- Returns:
Table data as
collections.OrderedDict
instances.- Return type:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.TableNotFoundError – If the table not found in the database.
simplesqlite.OperationalError – If failed to execute a query.
- Example:
- 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:
- Returns:
Table data as a
SimpleSQLite
instance that connected to in memory database.- Return type:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.TableNotFoundError – If the table not found in the database.
simplesqlite.OperationalError – If failed to execute a query.
- 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:
- Returns:
Table data as a
tabledata.TableData
instance.- Return type:
tabledata.TableData
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.TableNotFoundError – If the table not found in the database.
simplesqlite.OperationalError – If failed to execute a query.
Note
pandas
package required to execute this method.
- set_row_factory(row_factory: Callable | None) None [source]¶
Set row_factory to the database connection.
- update(table_name: str, set_query: str | Sequence[Set], where: str | Where | And | Or | None = None) Cursor | None [source]¶
Execute an UPDATE query.
- Parameters:
- Raises:
simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
simplesqlite.TableNotFoundError – If the table not found in the database.
simplesqlite.OperationalError – If failed to execute a query.
- validate_access_permission(valid_permissions: Sequence[str]) None [source]¶
- Parameters:
valid_permissions (
list
/tuple
) – List of permissions that access is allowed.- Raises:
ValueError – If the
mode
is invalid.simplesqlite.NullDatabaseConnectionError – If not connected to a SQLite database file.
- verify_attr_existence(table_name: str, attr_name: str) None [source]¶
- Parameters:
- Raises:
simplesqlite.AttributeNotFoundError – If attribute not found in the table
simplesqlite.TableNotFoundError – If the table not found in the database.
- 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