Manager

This module is responsible for managing synchronous database operations.

class DbUnify.SQLite3.sync.Manager.Manager.Manager(db_name: str, cache_ttl: int = 300)

Bases: object

# Manager Class:

#### The Manager class provides an interface for managing SQLite databases asynchronously. It offers methods for connecting to the database, executing SQL queries, creating and modifying tables, inserting and deleting rows, and closing the database connection.

### Attributes:

• db_name (str): The name of the SQLite database.

• raw (Raw): An instance of the Raw class for executing raw SQL queries.

• connection: The connection object to the SQLite database.

• cursor: The cursor object for executing SQL queries.

• cache (Cache): An instance of the Cache class for caching query results.

### Methods:

• __init__(self, db_name): Initializes the Manager instance with the name of the SQLite database.

• connect(self): Asynchronously connects to the SQLite database.

• fetch_all(self, query, *args): Executes a query and fetches all results.

• create_table(self, table_name, columns): Creates a table in the database.

• drop_table(self, table_name): Drops a table from the database.

• add_column(self, table_name, column_name, data_type): Adds a column to an existing table.

• insert_row(self, table_name, values): Inserts a row into the table.

• delete_column(self, table_name, column_name): Deletes a column from the table.

• delete_row(self, table_name, condition): Deletes a row from the table based on a condition.

• update_row(self, table_name, values, condition): Updates a row in the table based on a condition.

• select_one(self, table_name, condition): Searches for a single row in the table based on a condition.

• select(self, table_name): Searches for all rows in the table.

• close(self): Closes the database connection.

### Raises:

• ConnectionError: If there is an error connecting to or closing the database.

• RuntimeError: If there is an error executing SQL queries, creating or modifying tables, inserting or deleting rows, or searching for rows.

### Note:

• The ‘Raw’ class is used internally for executing raw SQL queries.

add_column(table_name: str, column_name: str, data_type: str, constraints: str) → None

Add a column to an existing table.

Parameters:

• table_name (str) – Name of the table to add the column to.

• column_name (str) – Name of the column to be added.

• data_type (str) – Data type of the column.

• constraints (str) – Constraints to be applied on the column.

Raises:

RuntimeError – If there is an error adding the column.

close() → None

Close the database connection.

Raises:

ConnectionError – If there is an error closing the connection.

connect() → None

Connect to the SQLite database.

create_table(table_name: str, columns: List[Tuple[str, str, List[str | Rules] | None]]) → None

Create a table in the database.

Parameters:

• table_name (str) – Name of the table to be created.

• columns (list) – List of tuples containing column names, data types, and rules.

Raises:

RuntimeError – If there is an error creating the table.

delete_column(table_name: str, column_name: str) → None

Delete a column from the table.

Parameters:

• table_name (str) – Name of the table.

• column_name (str) – Name of the column to be deleted.

Raises:

RuntimeError – If there is an error deleting the column.

delete_row(table_name: str, condition: str, *args) → None

Delete a row from the table based on a condition.

Parameters:

• table_name (str) – Name of the table.

• condition (str) – SQL condition to match rows to be deleted.

• *args – Parameters to be passed to the condition.

Raises:

RuntimeError – If there is an error deleting the row.

drop_table(table_name: str) → None

Drop a table from the database.

Parameters:

table_name (str) – Name of the table to be dropped.

Raises:

RuntimeError – If there is an error dropping the table.

fetch_all(query: str, *args) → List[Tuple]

Execute a query and fetch all results.

Parameters:

• query (str) – The SQL query to be executed.

• *args – Parameters to be passed to the query.

Returns:

List of fetched rows.

Return type:

list

Raises:

RuntimeError – If there is an error fetching data.

get_table_columns(table_name: str) → Dict[str, str]

Get the columns and their data types for a table.

Parameters:

table_name (str) – Name of the table.

Returns:

Dictionary of column names and their data types.

Return type:

dict

Raises:

RuntimeError – If there is an error getting the columns.

insert_row(table_name: str, values: Dict[str, str | int | float]) → None

Insert a row into the table.

Parameters:

• table_name (str) – Name of the table to insert the row into.

• values (dict) – Dictionary of column-value pairs for the row.

Raises:

RuntimeError – If there is an error inserting the row.

select(table_name: str) → List[Dict[str, str | int | float]]

Search for all rows in the table.

Parameters:

table_name (str) – Name of the table.

Returns:

List of rows, each as a dictionary of column-value pairs.

Return type:

list

Raises:

RuntimeError – If there is an error searching for rows.

select_one(table_name: str, condition: str, *args) → Dict[str, str | int | float] | None

Search for a single row in the table based on a condition.

Parameters:

• table_name (str) – Name of the table.

• condition (str) – SQL condition to match the row.

• *args – Parameters to be passed to the condition.

Returns:

The matched row as a dictionary of column-value pairs, or None if no row is found.

Return type:

dict

Raises:

RuntimeError – If there is an error searching for the row.

update_row(table_name: str, values: Dict[str, str | int | float], condition: str, *args) → None

Update a row in the table based on a condition.

Parameters:

• table_name (str) – Name of the table.

• values (dict) – Dictionary of column-value pairs for the update.

• condition (str) – SQL condition to match the row to be updated.

• *args – Parameters to be passed to the condition.

Raises:

RuntimeError – If there is an error updating the row.

Examples

add_column

Add a column to an existing table:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Add a new column to an existing table
manager.add_column(
    table_name='users',
    column_name='age',
    data_type='INTEGER',
    constraints='DEFAULT 18'
)

# Close the connection
manager.close()

close

Close the database connection:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Perform some database operations...

# Close the database connection
manager.close()

connect

Connect to the SQLite database:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the SQLite database
manager.connect()

create_table

Create a new table:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Create a new table
manager.create_table(
    table_name='users',
    columns=[
        ('id', 'INTEGER', ['PRIMARY KEY', 'AUTOINCREMENT']),
        ('name', 'TEXT', ['NOT NULL']),
        ('email', 'TEXT', ['UNIQUE'])
    ]
)

# Close the connection
manager.close()

delete_column

Delete a column from an existing table:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Delete a column from an existing table
manager.delete_column(
    table_name='users',
    column_name='age'
)

# Close the connection
manager.close()

delete_row

Delete a row based on a condition:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Delete a row from the table
manager.delete_row(
    table_name='users',
    condition='id = ?',
    1
)

# Close the connection
manager.close()

drop_table

Drop a table from the database:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Drop a table from the database
manager.drop_table('users')

# Close the connection
manager.close()

fetch_all

Execute a query and fetch all results:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Fetch all results for a query
results = manager.fetch_all('SELECT * FROM users')

# Print results
print(results)

# Close the connection
manager.close()

get_table_columns

Get the columns and their data types for a table:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Get columns and data types for a table
columns = manager.get_table_columns('users')

# Print columns
print(columns)

# Close the connection
manager.close()

insert_row

Insert a row into the table:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Insert a row into the table
manager.insert_row(
    table_name='users',
    values={
        'name': 'John Doe',
        'email': 'john.doe@example.com'
    }
)

# Close the connection
manager.close()

select

Search for all rows in the table:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Fetch all rows from the table
rows = manager.select('users')

# Print rows
print(rows)

# Close the connection
manager.close()

select_one

Search for a single row based on a condition:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Fetch a single row from the table
row = manager.select_one(
    table_name='users',
    condition='email = ?',
    'john.doe@example.com'
)

# Print row
print(row)

# Close the connection
manager.close()

update_row

Update a row in the table based on a condition:

# Create an instance of the Manager class
manager = Manager(db_name='example.db')

# Connect to the database
manager.connect()

# Update a row in the table
manager.update_row(
    table_name='users',
    values={
        'name': 'Jane Doe'
    },
    condition='email = ?',
    'john.doe@example.com'
)

# Close the connection
manager.close()