Skip to main content

DBActor: ORM helper and alternative

Project description


Simple class that interacts with Postgres to avoid ORM specifics.


DBActor supports multiple types of integrations. In many cases, you're looking to use it for a specific purpose. Sepcific extras installations allows for this.

Examples of what the different extras allows you to do are below.

$ pip install dbactor
$ pip install dbactor['jinjasql']
$ pip install dbactor['sqlalchemy']
$ pip install dbactor['pandas']
$ pip install dbactor['all']


Preference for DBActor location is in db/

import os
from dbactor import DBActor

db_name = os.environ.get('DB_NAME', 'dbname')
db_user = os.environ.get('DB_USER', 'dbuser')
db_password = os.environ.get('DB_PASSWORD', 'dbpassword')
db_host = os.environ.get('DB_HOST', 'localhost')
db_port = os.environ.get('DB_PORT', '5432')

actor = DBActor(database=db_name, user=db_user, password=db_password, host=db_host, port=db_port)

# Or initialize with the url
db_url = f'postgresql://{db_user}:{db_password}@{db_host}:{db_port}/{db_name}'

actor = DBActor(url=db_url)

Then, to use in any file import with,

from db import actor


In examples below, the following table.


__tablename__ == products

column_name data_type
id int4
name varchar
cost numeric

Statements for initial creation without using DBActor.

-- products.sql
-- Sequence and defined type

-- Table Definition
CREATE TABLE "public"."products" (
    "id" int4 NOT NULL DEFAULT nextval('products_id_seq'::regclass),
    "name" varchar,
    "cost" numeric,
    PRIMARY KEY ("id")

INSERT INTO "public"."products" ("id", "name", "cost") VALUES
('1', 'mug', '4'),
('2', 'fork', '1'),
('3', 'spoon', '1'),
('4', 'knife', '1'),
('5', 'pan', '8');


from db import actor

qstr = 'select count(*) from products'
res = actor.call_one(qstr)
# RealDictRow([('count', 5)])

qstr = 'select * from products;'

# If wanting RealDict
res = actor.call_all(qstr)
# [RealDictRow([('id', 1), ('name', 'mug'), ('cost', Decimal('4'))]), RealDictRow([('id', 2), ('name', 'fork'), ('cost', Decimal('1'))]), RealDictRow([('id', 3), ('name', 'spoon'), ('cost', Decimal('1'))]), RealDictRow([('id', 4), ('name', 'knife'), ('cost', Decimal('1'))]), RealDictRow([('id', 5), ('name', 'pan'), ('cost', Decimal('8'))])]

# If wanting python dict rather than RealDict:
res = actor.call_all_dict(qstr)
# [{'id': 1, 'name': 'mug', 'cost': Decimal('4')}, {'id': 2, 'name': 'fork', 'cost': Decimal('1')}, {'id': 3, 'name': 'spoon', 'cost': Decimal('1')}, {'id': 4, 'name': 'knife', 'cost': Decimal('1')}, {'id': 5, 'name': 'pan', 'cost': Decimal('8')}]


In cases where you may or may not have paramaters, JinjaSql is a great way to design queries, and DBActor allows for those in all cases.

from dbactor import DBJinjaSqlActor
from db import db_url

actor = DBJinjaSqlActor(url=db_url)

qstr = '''
        1 = 1
        {% if min_cost %}
        AND asdf >= min_cost
        {% endif %}
        {% if max_cost %}
        AND asdf <= max_cost
        {% endif %}

qparams = dict(min_cost=3, max_cost=4)
res = actor.call_all_dict(qstr, qparams=qparams)
# [{'id': 1, 'name': 'mug', 'cost': Decimal('4')}]


DBActor is made to avoid using ORMs for the most part. However, when inserting or updating rows, ORMs prove to be simpler. with DBSqlalchemyActor, you can either create or update with sql strings alone, or also with SQLAlchemy models.

# db/
from sqlalchemy import Column, Integer, Float, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Product(Base):
    __tablename__ = 'products'

    id = Column(Integer, primary_key=True)
    name = Column(Integer)
    price = Column(Float)

A perfect use case for this might be importing csvs.

Imagine a case where every day, someone creates a new csv with updated prices for the Things. In this case, there might be a daily job of looping through the csv and wanting to create or update an object for each row.


mug, 10
cup, 5
plate, 3
bowl, 6
import csv
from dbactor import DBSqlAlchemyActor
from db import db_url
from db.models import Product

actor = DBSqlAlchemyActor(url=db_url)

with open('products.csv', 'r') as csvfile:
    csvreader = csv.DictReader(csvfile)
    for row in csvreader:
        keys = dict(name=row['name'])
        values = dict(price=row['price'])
        actor.create_or_update_model(Product, keys=keys, values=values)
qstr = 'select * from products'
res = actor.call_all_dict(qstr)
# [{'id': 2, 'name': 'fork', 'cost': Decimal('1')}, {'id': 3, 'name': 'spoon', 'cost': Decimal('1')}, {'id': 4, 'name': 'knife', 'cost': Decimal('1')}, {'id': 5, 'name': 'pan', 'cost': Decimal('8')}, {'id': 1, 'name': 'mug', 'cost': Decimal('10')}, {'id': 6, 'name': 'cup', 'cost': Decimal('5')}, {'id': 7, 'name': 'plate', 'cost': Decimal('3')}, {'id': 8, 'name': 'bowl', 'cost': Decimal('6')}]

qstr = 'select count(*) from products'
res = actor.call_one(qstr)
# RealDictRow([('count', 8)])


When using DBPandasActor, you'll have the ability to call_df using the same connection. This is done with JinjaSQL templates.

from dbactor import DBPandasActor
from db import db_url

actor = DBPandasActor(url=db_url)

qstr = '''
select * from products where cost > {{min_cost}};
qparams = dict(min_cost=4)

df = actor.call_df(qstr, qparams=qparams)
# <class 'pandas.core.frame.DataFrame'>
#    id   name  cost
# 0   2   fork   1.0
# 1   3  spoon   1.0
# 2   4  knife   1.0
# 3   5    pan   8.0
# 4   1    mug  10.0
# 5   6    cup   5.0
# 6   7  plate   3.0
# 7   8   bowl   6.0

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

dbactor-0.0.3.tar.gz (7.0 kB view hashes)

Uploaded Source

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page