ddcdatabases 1.0.17

Databases Connection and Queries

Databases Connection and Queries

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 variables
• 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
• SYNC sessions defaults:
  • autoflush is True
  • expire_on_commit is True
  • echo is False
• ASYNC sessions defaults:
  • autoflush is True
  • expire_on_commit is False
  • echo is False

SQLITE

class Sqlite(
    filepath: Optional[str] = None,
    echo: Optional[bool] = None,
    autoflush: Optional[bool] = None,
    expire_on_commit: Optional[bool] = None,
    extra_engine_args: Optional[dict] = None,
)

Session

import sqlalchemy as sa
from ddcDatabases import DBUtils, Sqlite
with Sqlite() as session:
    utils = DBUtils(session)
    stmt = sa.select(TableModel).where(TableModel.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,
    user: 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,
    autoflush: Optional[bool] = None,
    expire_on_commit: Optional[bool] = None,
    extra_engine_args: Optional[dict] = None,
)

Sync Example

import sqlalchemy as sa
from ddcDatabases import DBUtils, MSSQL
with MSSQL() as session:
    stmt = sa.select(TableModel).where(TableModel.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(TableModel).where(TableModel.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,
    user: Optional[str] = None,
    password: Optional[str] = None,
    database: Optional[str] = None,
    echo: Optional[bool] = None,
    autoflush: Optional[bool] = None,
    expire_on_commit: Optional[bool] = None,
    engine_args: Optional[dict] = None,
)

Sync Example

import sqlalchemy as sa
from ddcDatabases import DBUtils, PostgreSQL
with PostgreSQL() as session:
    stmt = sa.select(TableModel).where(TableModel.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(TableModel).where(TableModel.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 using Poe

poetry update --with test
poe tests

License

Released under the MIT License

Buy me a cup of coffee

• GitHub Sponsor
• ko-fi
• Paypal