sqlalchemybulk 0.1.5

Perform CRUD operations with SQLAlchemy

Perform Create, Update, Read, and Delete operations easily using the SQLAlchemy orm

—

Installation

To install via pip:

pip install sqlalchemybulk

Or download the source code and install manually:

git clone https://github.com/sudonorm/sqlalchemy-bulk.git
cd sqlalchemy-bulk/
python -m pip install .

Or download the source code and use the setup.py file:

git clone https://github.com/sudonorm/sqlalchemy-bulk.git
cd sqlalchemy-bulk/
python setup.py install

Be sure to add a .env to the root of the project with at least these variables (DB_URI, ALEMBIC_PATH):

DB_URI=”postgresql+psycopg2://username:password@host:port/database_name” ALEMBIC_PATH=”<path/to/user/project/root/folder>”

“SQLITE_DB_PATH” is optional, except you are testing with a SQLite DB.

SQLITE_DB_PATH=”<path/to/user/folder>/test.db”

Basic usage

from sqlalchemybulk.crud_helper_funcs import UploadData, DownloadData, DeleteData
import pandas as pd
from sqlalchemy import select, delete
from tests.sample_data.dataModel import engine
from tests.sample_data import dataModel

## Insert (Create) or Update
upload_data = UploadData(engine=engine)
returned_ids = upload_data.upload_info_atomic(dbTable='dataModel.Address', df=df, unique_idx_elements=['name', 'postalZip'], column_update_fields=['address', 'country', 'suptext', 'numberrange', 'currency', 'alphanumeric'])

## Download (Read)
download_data = DownloadData(engine=engine)

### query full table
query = select(dataModel.Address)
result = download_data.download_info_using_session(statement=query)

### query with filter
query = select(dataModel.Address).where(dataModel.Address.postalZip == "3778")
result = download_data.download_info_using_session(statement=query)

## Delete
delete_data = DeleteData(engine=engine)
query = delete(dataModel.Address).where(dataModel.Address.postalZip == "15143")
delete_data.delete_data_on_condition(dbTable="dataModel.Address", statement=query)

Bugs, requests, questions, etc.

Please create an issue on GitHub.