Insert and update with core SQLAlchemy

Question

I have a database that I don't have metadata or orm classes for (the database already exists).

I managed to get the select stuff working by:

from sqlalchemy.sql.expression import ColumnClause
from sqlalchemy.sql import table, column, select, update, insert
from sqlalchemy.ext.declarative import *
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
import pyodbc

db = create_engine('mssql+pyodbc://pytest')
Session = sessionmaker(bind=db)
session = Session()

list = []
list.append (column("field1"))
list.append (column("field2"))
list.append (column("field3"))

s = select(list)
s.append_from('table')
s.append_whereclause("field1 = 'abc'")
s = s.limit(10)

result = session.execute(s)
out = result.fetchall()

print(out)

So far so good.

The only way I can get an update/insert working is by executing a raw query like:

session.execute(<Some sql>)

I would like to make it so I can make a class out of that like:

u = Update("table")
u.Set("file1","some value")
u.Where(<some conditon>)

seasion.execute(u)

Tried (this is just one of the approaches I tried):

i = insert("table")
v = i.values([{"name":"name1"}, {"name":"name2"}])

u = update("table")
u = u.values({"name": "test1"})

I can't get that to execute on:

session.execute(i)

or

session.execute(u)

Any suggestion how to construct an insert or update without writing ORM models?

Answer 1

As you can see from the SQLAlchemy Overview documentation, sqlalchemy is build with two layers: ORM and Core. Currently you are using only some constructs of the Core and building everything manually.

In order to use Core you should let SQLAlchemy know some meta information about your database in order for it to operate on it. Assuming you have a table mytable with columns field1, field2, field3 and a defined primary key, the code below should perform all the tasks you need:

from sqlalchemy.sql import table, column, select, update, insert

# define meta information
metadata = MetaData(bind=engine)
mytable = Table('mytable', metadata, autoload=True)

# select
s = mytable.select() # or:
#s = select([mytable]) # or (if only certain columns):
#s = select([mytable.c.field1, mytable.c.field2, mytable.c.field3])
s = s.where(mytable.c.field1 == 'abc')
result = session.execute(s)
out = result.fetchall()
print(out)

# insert
i = insert(mytable)
i = i.values({"field1": "value1", "field2": "value2"})
session.execute(i)

# update
u = update(mytable)
u = u.values({"field3": "new_value"})
u = u.where(mytable.c.id == 33)
session.execute(u)