The ORM in SQLAlchemy is helpful for projects with complex relations and logic, but when there are only simple data types SQLAlchemy Core is often all you need. I have a few pieces of data that map to a table easily, and relations aren’t important. For that case, the humble collections.namedtuple is invaluable. Declaring a namedtuple that matches column names is an easy way to add a little syntax sugar on top of SQLAlchemy Core without using the ORM.,Simple, and making a user looks a lot like if User were a normal object. Now let’s build the persistence portion.,This isn’t so great, remembering indices is easy to get wrong. Worse, it can be the source of really annoying (and sometimes sneaky) errors down the road if the data in multiple columns looks similar. For us, mixing up email addresses and names will be easy to notice since most people don’t have “.com” in their surname.,Note that some of the tuples are the User namedtuple, while others are raw tuples. As long as the data in the tuples match the schema, SQLAlchemy will save either properly.
Before we get to that, let’s set up a data model. We’ll just track user names
and email addresses, so the namedtuple will look like this:
from collections
import namedtuple
User = namedtuple('User', ["id", "name", "email"])
someone = User(1, "Some Body", "some@one.com")import json, sqlalchemy
connection_string = 'sqlite://'
db = sqlalchemy.create_engine(connection_string)
engine = db.connect()
meta = sqlalchemy.MetaData(engine)
columns = (
sqlalchemy.Column('id', sqlalchemy.Integer),
sqlalchemy.Column('name', sqlalchemy.Text),
sqlalchemy.Column('email', sqlalchemy.Text),
)
sqlalchemy.Table("users", meta, * columns)
meta.create_all()
table = sqlalchemy.table("users", * columns)# add test data
statements = [
table.insert().values(user)
for user in (
User(1, "Alice", "alice@test.com"),
User(2, "Bob", "bob@test.com"),
(3, "Chuck", "Chuck@test.com"),
(4, "Diane", "diane@test.com"),
)
]
[engine.execute(stmt) for stmt in statements]With namedtuple, accessing data and even debugging becomes easier.
tuple_alice = User( * alice)
print("User: {}".format(tuple_alice.name))
# User: Alice
print(alice)
#(1, 'Alice', 'alice@test.com')
print(tuple_alice)
# User(id = 1, name = 'Alice', email = 'alice@test.com')import json, sqlalchemy
connection_string = 'sqlite://'
db = sqlalchemy.create_engine(connection_string)
engine = db.connect()
meta = sqlalchemy.MetaData(engine)
columns = (
sqlalchemy.Column('id', sqlalchemy.Integer),
sqlalchemy.Column('name', sqlalchemy.Text),
sqlalchemy.Column('email', sqlalchemy.Text),
)
sqlalchemy.Table("users", meta, * columns)
meta.create_all()
table = sqlalchemy.table("users", * columns)# add test data
statements = [
table.insert().values(user)
for user in (
User(1, "Alice", "alice@test.com"),
User(2, "Bob", "bob@test.com"),
(3, "Chuck", "Chuck@test.com"),
(4, "Diane", "diane@test.com"),
)
]
[engine.execute(stmt) for stmt in statements]Namedtuples have a bunch of behaviors that make them unsuitable for mapping with sqlalchemy: Most importantly, namedtuples can't be changed after they are created. This means that you can't use a namedtuple to track the state of a row in the database after the say an insert operation. You would typically want do something like this:
class MyDataHolder(namedtuple('MyDataHolder', ('id', 'my_value')):
pass
mapper(MyDataHolder, MyDataMeta)
...
newRow = MyDataHolder(None, 'AAA')
...
session.add(newRow)The mapper seems to add a _init_method. So doing the following after the mapper statement makes it work again:
del Point.__init__Hello, I have been trying to get a namedtuple to work with SQLalchemy, but to no avail.. Web search hasn't been very illuminating and I'm new with Python and SQLalchemy so I'm not really sure if I'm chasing windmills :( The basic idea is that I have a namedtuple, ie:,Any ideas why that happens? Does anyone know how to make a namedtuple work with sqlalchemy? Of course I can define my own Point class, but I'm obsessing over making namedtuple work now..,is not going to work with sqlalchemy (I think). I can create a Bunch class but I won't know beforehand how many ints I want to store in my collection.. I will set it before I create my database. I hope I'm making sense..,The main idea is that I want a named collection of stuff that can be easily stored in a database. So the very pythonic:
Hello, I have been trying to get a namedtuple to work with SQLalchemy, but to no avail.. Web search hasn't been very illuminating and I'm new with Python and SQLalchemy so I'm not really sure if I'm chasing windmills :( The basic idea is that I have a namedtuple, ie:
>> Point = namedtuple('Point', ['x', 'y'])which basically creates a class Point(tuple) if I'm correct. This works fine at first and I can create objects like:
>> p = Point(3, 4)
But after I create the engine etc and call mapper , I can't create any more objects without getting this error:
Traceback (most recent call last):
File "<pyshell#62>", line 1, in <module>
f=Point(3,4)
TypeError: __init__() takes exactly 1 argument (3 given)but I can't add them to the base, without getting
Class '__main__.Point'
is mapped, but this instance lacks
instrumentation.This occurs when the instance is created before
sqlalchemy.orm.mapper(__main__.Point) was called.I'm trying something like this:
from sqlalchemy
import *
from sqlalchemy.orm
import *
from collections
import namedtuple
Point = namedtuple('Point', ['x', 'y'], verbose = True)
p = Point(3, 4)
db = create_engine('sqlite:///pointtest.db')
metadata = MetaData()
pointxy = Table('pointxy', metadata,
Column('no', Integer, primary_key = True),
Column('x', Integer),
Column('y', Integer),
sqlite_autoincrement = True)
metadata.create_all(db)
m = mapper(Point, pointxy)
Session = sessionmaker(bind = db)
session = Session()
f = Point(3, 4)The main idea is that I want a named collection of stuff that can be easily stored in a database. So the very pythonic:
class Bunch:
__init__ = lambda self, ** kw: setattr(self, '__dict__', kw)SQLAlchemy now supports Python asyncio-compatible database drivers using an all-new asyncio front-end interface to Connection for Core usage as well as Session for ORM use, using the AsyncConnection and AsyncSession objects.,The user facing async API itself is focused around IO-oriented methods such as AsyncEngine.connect() and AsyncConnection.execute(). The new Core constructs strictly support 2.0 style usage only; which means all statements must be invoked given a connection object, in this case AsyncConnection.,The initial database API supported is the asyncpg asyncio driver for PostgreSQL.,The error message is described in the errors page at This connection is on an inactive transaction. Please rollback() fully before proceeding.
with Session(engine, future = True) as sess:
stmt = select(User).where(
User.name == 'sandy'
).join(User.addresses).where(Address.email_address.like("%gmail%"))
result = sess.execute(stmt)
for user in result.scalars():
print(user)session.query(User).filter(User.name == 'sandy').update({
"password": "foobar"
}, synchronize_session = "fetch")with Session(engine, future = True) as sess:
stmt = update(User).where(
User.name == 'sandy'
).values(password = "foobar").execution_options(
synchronize_session = "fetch"
)
sess.execute(stmt)session = Session(bind = engine) for id_ in random.sample(ids, n): result = session.query(Customer).filter(Customer.id == id_).one()
test_orm_query: (10000 iterations);
total time 3.440652 sectest_orm_query: (10000 iterations);
total time 2.367934 sec