Installation

Install Use Pypi

pip install DbUnify



Install Use Github

git clone https://github.com/Sepehr0Day/DbUnify.git




Connect To Database

for easy use we use sync classes but code not different with aio(async) version, its same.


Code:


import DbUnify.sync as sync

# Initialize a database manager with the name 'database.db'
db_manager = sync.Manager('database.db')

Output:


None

Class Manager

Functions

create_table

Code:

Parameters:

  • table_name (str): Name of the table to be created.
  • columns (List): List of tuples containing column names and data types.

db_manager = sync.Manager('database.db')

# Create a table named 'employees' with specified columns
db_manager.create_table('employees', [
    ('id', 'INTEGER PRIMARY KEY'), 
    ('name', 'TEXT'),
    ('age', 'INTEGER'),
    ('department', 'TEXT')
    ])

Output:


None

drop_table

Code:

Parameters:

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

db_manager = sync.Manager('database.db')

# Drop the 'employees' table
db_manager.drop_table('employees')

Output:


None

add_column

Code:

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.

db_manager = sync.Manager('database.db')

# Add a new column 'description' to the 'department' table
db_manager.add_column('department', "description", "text")

Output:


None

insert_row

Code:

Parameters:

  • table_name (str): Name of the table to insert the row into.
  • values (dict): Dictionary of column-value pairs for the row.

db_manager = sync.Manager('database.db')

# Insert rows into the 'employees' table
db_manager.insert_row('employees', 
    {
        'name': 'Alice',
        'age': 30,
        'department': 'HR'
    })

Output:


None

delete_column

Code:

Parameters:

  • table_name (str): Name of the table to delete the column from.
  • values (str): Name of the column to be deleted.

db_manager = sync.Manager('database.db')

# Delete the column 'description' from the 'department' table
db_manager.delete_column('department', 'description')

Output:


None

delete_row

Code:

Parameters:

  • table_name (str): Name of the table to delete the row from.
  • condition (str): Condition for row deletion.

db_manager = sync.Manager('database.db')

# Delete a row from the 'employees' table where id = 1
db_manager.delete_row('employees', 'id = 2')

Output:


None

update_row

Code:

Parameters:

  • table_name (str): Name of the table to update the row in.
  • values (dict): Dictionary of column-value pairs to be updated.
  • condition (str): Condition for row update.

db_manager = sync.Manager('database.db')

# Update a row in the 'employees' table where id = 1
db_manager.update_row('employees', 
    {
        'name': 'Alice',
        'age': 31,
        'department': 'HR'
    },
        'id = 1')

Output:


None

select_one

Code:

Parameters:

  • table_name (str): Name of the table to search in.
  • condition (str): Condition for row search.

db_manager = sync.Manager('database.db')

# Select a row from the 'employees' table where id = 1
data = db_manager.select_one('employees', 'id = 1')
print(data)

Output:


(1, 'Alice', 30, 'HR')

select

Code:

Parameters:

  • table_name (str): Name of the table to search in.

db_manager = sync.Manager('database.db')

# Select all rows from the 'employees' table
employees = db_manager.select('employees')
print(employees)

Output:


[(1, 'Alice', 30, 'HR'), (3, 'Charlie', 40, 'Finance')]

fetch_all

Code:

Parameters:

  • query (str): The SQL query to be executed and fetchall.
  • *args (Any): Parameters to be passed to the query.

db_manager = sync.Manager('database.db')

# Fetch information about the 'employees' table
db_manager.fetch_all("PRAGMA table_info('employees')")

Output:


[(0, 'id', 'INTEGER', 0, None, 1), 
(1, 'name', 'TEXT', 0, None, 0),
(2, 'age', 'INTEGER', 0, None, 0),
(3, 'department', 'TEXT', 0, None, 0)]

close

Code:

Raises:

  • ConnectionError : If there is an error closing the connection.

db_manager = sync.Manager('database.db')

db_manager.close()

Output:


Noen




Class Exporter

Functions



export_chart_database

Code:

Parameters:

  • output_directory (str): The directory where chart images will be saved.
  • chart_type (str): The type of chart to create ('bar', 'line', 'scatter', 'histogram').
  • x_label (str): Label for the x-axis.
  • y_label (str): Label for the y-axis.

db_manager = sync.Manager('database.db')
exporter = sync.Exporter(db_manager)

# Export the entire database to a chart
path = '.'
exporter.export_chart_database(path, 'line', 'X', 'Y')

Output:


None

export_chart_table

Code:

Parameters:

  • table_name (str): Name of the database table to retrieve data from.
  • x_column (int): Index of the x-axis column in the retrieved data.
  • y_column (int): Index of the y-axis column in the retrieved data.
  • x_label (str): Label for the x-axis.
  • y_label (str): Label for the y-axis.
  • title (str): Title of the chart.
  • save_path (str): Path to save the chart image.
  • chart_type (str): The type of chart to create ('bar', 'line', 'scatter', 'histogram')

db_manager = sync.Manager('database.db')
exporter = sync.Exporter(db_manager)

# Export data from the 'employees' table to a chart
exporter.export_chart_table(
    table_name='employees',
    save_path='employees_chart.png',
    y_column=2, 
    x_column=1,  
    y_label='age',
    x_label='name'
)

Output:


None

export_data_csv

Code:

Parameters:

  • table_name (str): The name of the table to export data from.
  • csv_file_path (str): The path to save the exported CSV file.

db_manager = sync.Manager('database.db')
exporter = sync.Exporter(db_manager)

# Export data from the 'employees' table to a CSV file
exporter.export_data_csv('employees', path)

Output:


None


Class Raw

Functions

execute_query

Code:

Parameters:

  • query (str): The SQL query to be executed.
  • *args (str): Parameters to be passed to the query.

db_manager = sync.Manager('database.db')
raw = sync.Raw(db_manager)

# Execute a query to delete a row from the 'department' table where id = 1
data = raw.execute_query("DELETE FROM department WHERE id = 1;")
print(data)

Output:


True

list_tables

Code:


db_manager = sync.Manager('database.db')
raw = sync.Raw(db_manager)

# List all tables in the database
tables = raw.list_tables()
print(tables)

Output:


['employees']

insert_base64

Code:

Parameters:

  • table_name (str): Name of the table to insert data into.
  • data_dict (dict): A dictionary where keys are column names, and values are data to be encoded and inserted.

db_manager = sync.Manager('database.db')
raw = sync.Raw(db_manager)

# Insert a row into the 'employees' table using base64 encoding
raw.insert_base64('employees', 
    {
        'name': 'max',
        'age': 24,
        'department': 'security'
    })

Output:


None

read_base64

Code:

Parameters:

  • table_name (str): Name of the table to read data from.
  • data_dict (dict): If True, only return rows where at least one column contains base64 encoded data.

db_manager = sync.Manager('database.db')
raw = sync.Raw(db_manager)

# Read data from the 'employees' table using base64 encoding
data = raw.read_base64('employees', True)
print("Output only_base64=True : \n" + data)

data = raw.read_base64('employees', False)
print("Output only_base64=False : \n" + data)

Output:


Outout only_base64=True : 
[{'id': 4, 'name': b'max', 'age': b'24', 'department': b'security'}]

Output only_base64=False :  
[{'id': 1, 'name': 'Alice', 'age': 30, 'department': 'HR'},
{'id': 2, 'name': 'Bob', 'age': 35, 'department': 'IT'},
{'id': 3, 'name': 'Charlie', 'age': 40, 'department': 'Finance'},
{'id': 4, 'name': b'max', 'age': b'24', 'department': b'security'}]

backup_database

Code:

Parameters:

  • backup_path (str): The path where the backup should be stored.

db_manager = sync.Manager('database.db')
raw = sync.Raw(db_manager)

# Backup the database to a file named 'backup.db'
raw.backup_database('C://backup.db')

Output:


True

backup_database

Code:

Parameters:

  • backup_path (str): The path where the backup should be stored.

db_manager = sync.Manager('database.db')
raw = sync.Raw(db_manager)

# Restore the database from the backup file 'backup.db'
raw.restore_database('C://backup.db')

Output:


True


Issues

If you have any issue please raise a PR in github