zschumacher / pydapper

a pure python port of the NuGet library dapper

pydapper

A pure python library inspired by the NuGet library dapper.

pydapper is built on top of the dbapi 2.0 spec to provide more convenient methods for working with databases in python.

Help

See the documentation for more details.

Installation

It is recommended to only install the database apis you need for your use case. Example below is for psycopg2!

pip

pip install pydapper[psycopg2]

poetry

poetry add pydapper -E psycopg2

Never write this again...

from psycopg2 import connect

@dataclass
class Task:
    id: int
    description: str
    due_date: datetime.date

with connect("postgresql://pydapper:pydapper@localhost/pydapper") as conn:
    with conn.cursor() as cursor:
        cursor.execute("select id, description, due_date from task")
        headers = [i[0] for i in cursor.description]
        data = cursor.fetchall()

list_data = [Task(**dict(zip(headers, row))) for row in data]

Instead, write...

from dataclasses import dataclass
import datetime

import pydapper


@dataclass
class Task:
    id: int
    description: str
    due_date: datetime.date

    
with pydapper.connect("postgresql+psycopg2://pydapper:pydapper@locahost/pydapper") as commands:
    tasks = commands.query("select id, description, due_date from task;", model=Task)

(This script is complete, it should run "as is")

Buy me a coffee

If you find this project useful, consider buying me a coffee!

pydapper's Issues

explore some tools for managing changelogs and releases

• https://pypi.org/project/gitchangelog/
• https://commitizen-tools.github.io/commitizen/
• https://github.com/samuelcolvin/pydantic/tree/master/changes

SSL support

Please correct me if i am wrong, but i cannot find any way to specify the ssl parameters when using pydapper

Add support for new oracledb package

It looks like Oracle is changing their driver package to be (mostly) pure Python, and is changing some of the interfaces.

Here's a link to some documentation that shows all of the differences:
https://python-oracledb.readthedocs.io/en/latest/user_guide/appendix_c.html

How to document a CASCADE

Maybe this is a dumb question.

I see the ‘Serialize a one-to-one relationship’. How would I mark it not only as a foreign key, but, eg SQLite “On UPDATE CASCADE”?

thank you!

add a `buffered` kwarg to query

if buffered is set to false, we should use fetchone via a generator to limit memory consumption

Returning data from "RETURNING" clause of INSERT

Using SQLite, i want to send a group of objects into the database and receive the inserted ones while the existing ones are ignored.

For than, i can use "... ON CONFLICT ... DO NOTHING RETURNING ...".

I've tried to do it with pydapper but i can't seem to get it to work. The data gets inserted but execute always returns zero.

Am i doing something wrong or is this unsupported by pydapper?
It seems to be unsupported but i may be wrong, hence my question.

Example code below:

import pydapper
from dataclasses import dataclass

@dataclass
class Person:
	id: int
	name: str
	address: str
	phone: int
	
	def __init__(self, name, address, phone):
		self.name = name
		self.address = address
		self.phone = phone

def do_something():
	people = [Person('Alice', 'Here', 1111), Person('Bob', 'There', 2222)]
	with pydapper.connect("sqlite://example.sqlite3") as commands:
		new_records = commands.execute(
			"INSERT INTO PEOPLE (name, address, phone) values (?name?, ?address?, ?phone?) ON CONFLICT(phone) DO NOTHING RETURNING *",
			param=people,
		)
		print(new_records)

Custom column mapping

I have an existing DB and want to try pydapper - since I already use Dapper.NET in C#.
But how to custom map columns (and, if possible, ignore) columns on classes?

Reimplement the query method

The query method can be optimized such that its not looping over a result set twice. This should give significant speed updates for larger queries.

How can I serialize a one to many relationship?

I do not find how it is possible to relate 1 to many relationship. is this possible?

🔧 Speed up CI

• add cacheing for oracle instant client install
• consider splitting up db tests into their own workflows
• cache poetry install

Shortening Code

Seems to me you still leave it to the caller to create a cursor for every query. I like to shorten this with a generator function like (taken from here):

def db_iter(conn, cmd, mapfn = lambda x : x) :
    "executes cmd on a new cursor from connection conn and" \
    " yields the results in turn."
    for item in conn.cursor().execute(cmd) :
        yield mapfn(item)
    #end for
#end db_iter

Then queries become as simple as

for row in db_iter(conn, "select ...") :
    ... do something with row ...
#end for

Also, instead of providing “buffered” versus “unbuffered” versions of queries, why not just provide the iterator (“unbuffered”) version; then returning the entire query list is as simple as

buffered_result = list(unbuffered_query())

Finally, I have a technique for providing both synchronous and asynchronous variants of an API with minimal code duplication. You can see it in action here.

✨ add async support

some popular async dbapis include:

• https://aiopg.readthedocs.io/en/stable/
• https://cx-oracle-async.readthedocs.io/en/latest/ (maybe)
• https://aiomysql.readthedocs.io/en/latest/
• https://github.com/aio-libs/aioodbc for mssql?

Add Sybase connection

New feature add easy sybase connections

✨ Add first class support for sprocs

A little more samples please
I'm struggling to call MSSQL stored procedure(s) with parameter(s)
like
create or alter procedure sp_TestCall( @foo int, @bar nvarchar(10) ) AS
BEGIN
select
@foo + 1 as Id,
GETDATE() as ServerTime
END

as dictionary-as-a-paramter-list ( my expectation is from the original .NET based dapper an anonymous type as-a-parameter-list ) :
...
with pydapper.connect( "mssql://bunny:[email protected]:1433/CarrotMSDB") as commands:
# commands.query_first( "sp_TestCall", param= { "foo": 1, "bar": U"some text" } )
# commands.query_first( "execute sp_TestCall", param= { "foo": 1, "bar": U"some text" } )
# commands.query_first( "execute sp_TestCall @foo, @bar", param= { "@foo": 1, "@bar": U"some text" } )
# commands.query_first( "execute sp_TestCall @foo, @bar", param= { "foo": 1, "bar": U"some text" } )

# commands.query_first( "execute sp_TestCall", param = dict( foo= 1, bar= U"some text" )  )

and many more combinations of the above...

always ends with:

Traceback (most recent call last):
File "/usr/local/lib/python3.7/dist-packages/pydapper/utils.py", line 11, in safe_getattr
return obj[key]
KeyError: ''

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/home/pi/Desktop/QueryValidCards.py", line 240, in CardPass
result = commands.query_first( "execute sp_TestCall", param= { "foo": 1, "bar": U"some text" } )
File "/usr/local/lib/python3.7/dist-packages/pydapper/commands.py", line 188, in query_first
handler.execute(cursor)
File "/usr/local/lib/python3.7/dist-packages/pydapper/commands.py", line 83, in execute
cursor.execute(self.prepared_sql, self.ordered_param_values)
File "/usr/local/lib/python3.7/dist-packages/cached_property.py", line 36, in get
value = obj.dict[self.func.name] = self.func(obj)
File "/usr/local/lib/python3.7/dist-packages/pydapper/commands.py", line 75, in prepared_sql
return pattern.sub(sub_param_with_placeholder, self._sql) # type: ignore
File "/usr/local/lib/python3.7/dist-packages/pydapper/commands.py", line 73, in sub_param_with_placeholder
return self.get_param_placeholder(matched_param_name)
File "/usr/local/lib/python3.7/dist-packages/pydapper/mssql/pymssql.py", line 26, in get_param_placeholder
param_value = safe_getattr(test_param, param_name)
File "/usr/local/lib/python3.7/dist-packages/pydapper/utils.py", line 16, in safe_getattr
raise KeyError(f"Key {key!r} can not be accessed on {obj!r} or does not exist")
KeyError: "Key '' can not be accessed on {'foo': 1, 'bar': 'some text'} or does not exist"

how to connect to sql server using windows authentication

Hi there, I'm a big dapper fan so good to see this project in Python!

I need to use windows auth to connect to a database server on the coporate network. The problem is that I'm not finding anywhere that pymssql supports this? I've tried to use pyodbc as per the below but that doesn't seem to work. Any help much appreciated, I'm new to Python.

Doc enhancements

• document using examples
• document connect examples

WARNING: pydapper 0.1.1a2 does not provide the extra 'mysql-connector-python'

Running the pip install command I get the warning WARNING: pydapper 0.1.1a2 does not provide the extra 'mysql-connector-python'. Not sure what that means new to python. When I take you example code and put it in to run I get "Could not derive dbapi from schemes ['mysql']" when it hits the statement below

with pydapper.connect(app.config["DB_CON_SITE_CONFIG"], autocommit=True) as commands:

Using this as the connection string 'mysql://myuser:mypasword:3306@localhost/mydb'

To make sure it wasn't the fact I couldn't get acccess to mysql and the database I tried this simple command entering the same information as the connection string and it worked.
with connect(
host="localhost",
user=input("Enter username: "),
password=getpass("Enter password: "),
database="mydb",
) as connection:
print(connection)
except Error as e:
print(e)

Coming from .net land where I constantly used dapper.net love to get this working.

Fix black dependency

Running black causes this error:

ImportError: cannot import name '_unicodefun' from 'click'

Pretty well known bug and is easily fixable. See more here: psf/black#2964

Support Python 3.12

It's been about 6 months since initial release, would like support for this

write the gha config file

• build and cacheing
• database setup
• Mypy, isort, black
• test and push to codecov
• build and push to pypi on releases

Add documentation that pymssql requires an explicit commit

I'm testing this with pymssql against a MSSQL instance. Following the examples given by the documentation you'd assume this was the way forward

from pydapper import connect
import os
from dataclasses import dataclass 

@dataclass
class Table:
    Name: str
    Description: str

test_model = Table("Test", "Test")

conn_string = os.getenv("connection_string")

with connect(conn_string) as commands:
    rows = commands.execute("INSERT INTO dbo.Table (Name, Description) VALUES (?Name?, ?Description?)", param=test_model.__dict__)
    assert rows == 1 # This is true

rows will always be 1 here, but the results are never committed to the database. If, however, you do:

with connect(conn_string) as commands:
    rows = commands.execute("INSERT INTO dbo.Table (Name, Description) VALUES (?Name?, ?Description?)", param=test_model.__dict__)
    assert rows == 1 # This is true
    commands.connection.commit()

Now the item will be written to the database.

If memory serves, I've seen the first approach work with SQLite in some earlier testing I did, so I'm not sure what the way forwards is:

Either this should be added to Commands.execute like so:

    def execute(self, sql: str, param: Union["ParamType", "ListParamType"] = None) -> int:
        handler = self.SqlParamHandler(sql, param)
        with self.cursor() as cursor:
            rowcount = handler.execute(cursor)
        self.connection.commit()
        return rowcount

Or the documentation should be updated to reflect the necessity of committing yourself. I'll happily submit a PR either way, but I'd be interested in your opinion on this first.

Deprecate cxOracle in favor of `oracledb`

re #24 and pending #25

🔧 Better error messages when params in your query don't match param input

Right now this is manifesting as a KeyError, but its not always obvious what exactly is happening for new users of the lib