pip install ddcDatabasesclass DBSqlite(db_file_path: str, batch_size=100, echo=False, future=True)
import sqlalchemy as sa
from ddcDatabases import DBSqlite, DBUtils
dbsqlite = DBSqlite(database_file_path)
with dbsqlite.session() as session:
stmt = sa.select(Table).where(Table.id == 1)
db_utils = DBUtils(session)
results = db_utils.fetchall(stmt)- Using driver "psycopg2" as default
class DBPostgres(future=True, echo=False, drivername="psycopg2", **kwargs)
import sqlalchemy as sa
from ddcDatabases import DBPostgres, DBUtils
db_configs = {
"username": username,
"password": password,
"host": host,
"port": port,
"database": database
}
dbpostgres = DBPostgres(**db_configs)
with dbpostgres.session() as session:
stmt = sa.select(Table).where(Table.id == 1)
db_utils = DBUtils(session)
results = db_utils.fetchall(stmt)- DBUTILS
- Uses SQLAlchemy statements
from ddcDatabases import DBUtils
db_utils = DBUtils(session)
db_utils.add(stmt)
db_utils.execute(stmt)
db_utils.fetchall(stmt)
db_utils.fetchone(stmt)
db_utils.fetch_value(stmt)- Using driver "asyncpg"
class DBPostgresAsync(future=True, echo=False, drivername="asyncpg", **kwargs)
import sqlalchemy as sa
from ddcDatabases import DBPostgresAsync, DBUtilsAsync
db_configs = {
"username": username,
"password": password,
"host": host,
"port": port,
"database": database
}
dbpostgres = DBPostgresAsync(**db_configs)
async with dbpostgres.session() as session:
stmt = sa.select(Table).where(Table.id == 1)
db_utils = DBUtilsAsync(session)
results = await db_utils.fetchall(stmt)- DBUTILS ASYNC
- Uses SQLAlchemy statements
from ddcDatabases import DBUtilsAsync
db_utils = DBUtilsAsync(session)
await db_utils.add(stmt)
await db_utils.execute(stmt)
await db_utils.fetchall(stmt)
await db_utils.fetchone(stmt)
await db_utils.fetch_value(stmt)poetry buildpoe testpoe coverageReleased under the MIT License
