ddcdatabases 1.0.11

Custom Database Connection and Queries

Few Utility Functions

Install All databases dependencies

pip install ddcDatabases[all]

Install MSSQL

pip install ddcDatabases[mssql]

Install PostgreSQL

pip install ddcDatabases[pgsql]

Databases

• Parameters for all classes are declared as OPTIONAL falling back to .env file
• All examples are using db_utils.py
• By default, the MSSQL class will open a session to the database, but the engine can be available

SQLITE

class Sqlite(
    file_path: Optional[str] = None,
    echo: Optional[bool] = None,
)

Session

import sqlalchemy as sa
from ddcDatabases import DBUtils, Sqlite
with Sqlite() as session:
    utils = DBUtils(session)
    stmt = sa.select(Table).where(Table.id == 1)
    results = utils.fetchall(stmt)
    for row in results:
        print(row)

Sync Engine

from ddcDatabases import Sqlite
with Sqlite().engine() as engine:
    ...

MSSQL

class MSSQL(        
    host: Optional[str] = None,
    port: Optional[int] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
    database: Optional[str] = None,
    schema: Optional[str] = None,
    echo: Optional[bool] = None,
    pool_size: Optional[int] = None,
    max_overflow: Optional[int] = None
)

Sync Example

import sqlalchemy as sa
from ddcDatabases import DBUtils, MSSQL
with MSSQL() as session:
    stmt = sa.select(Table).where(Table.id == 1)
    db_utils = DBUtils(session)
    results = db_utils.fetchall(stmt)
    for row in results:
        print(row)

Async Example

import sqlalchemy as sa
from ddcDatabases import DBUtilsAsync, MSSQL
async with MSSQL() as session:
    stmt = sa.select(Table).where(Table.id == 1)
    db_utils = DBUtilsAsync(session)
    results = await db_utils.fetchall(stmt)
    for row in results:
        print(row)

Sync Engine

from ddcDatabases import MSSQL
with MSSQL().engine() as engine:
    ...

Async Engine

from ddcDatabases import MSSQL
async with MSSQL().async_engine() as engine:
    ...

PostgreSQL

• Using driver psycopg2 as default

class DBPostgres(
    host: Optional[str] = None,
    port: Optional[int] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
    database: Optional[str] = None,
    echo: Optional[bool] = None,
)

Sync Example

import sqlalchemy as sa
from ddcDatabases import DBUtils, PostgreSQL
with PostgreSQL() as session:
    stmt = sa.select(Table).where(Table.id == 1)
    db_utils = DBUtils(session)
    results = db_utils.fetchall(stmt)
    for row in results:
        print(row)

Async Example

import sqlalchemy as sa
from ddcDatabases import DBUtilsAsync, PostgreSQL
async with PostgreSQL() as session:
    stmt = sa.select(Table).where(Table.id == 1)
    db_utils = DBUtilsAsync(session)
    results = await db_utils.fetchall(stmt)
    for row in results:
        print(row)

Sync Engine

from ddcDatabases import PostgreSQL
with PostgreSQL().engine() as engine:
    ...

Async Engine

from ddcDatabases import PostgreSQL
async with PostgreSQL().async_engine() as engine:
    ...

DBUtils and DBUtilsAsync

• Take an open session as parameter
• Can use SQLAlchemy statements
• Execute function can be used to update, insert or any SQLAlchemy.text

from ddcDatabases import DBUtils
db_utils = DBUtils(session)
db_utils.fetchall(stmt)                     # returns a list of RowMapping
db_utils.fetchvalue(stmt)                   # fetch a single value, returning as string
db_utils.insert(stmt)                       # insert into model table
db_utils.deleteall(model)                   # delete all records from model
db_utils.insertbulk(model, list[dict])      # insert records into model from a list of dicts
db_utils.execute(stmt)                      # this is the actual execute from session

Source Code

Build

poetry build -f wheel

Run Tests and Get Coverage Report

poetry run coverage run --omit=./tests/* --source=./ddcDatabases -m pytest -v && poetry run coverage report

License

Released under the MIT License

Buy me a cup of coffee

• GitHub Sponsor
• ko-fi
• Paypal