Python SqlAlchemy generate table classes pragmatically

Question

Given a list of table names, I'm trying to generate SqlAlchemy db table classes programmatically. Here is what I do:

def generate_table_class_list(engine):
    base = declarative_base(engine)
    table_names = ['table_a', 'table_b']
    table_classes = dict()
    for table in table_names:
        table_classes[table] = generate_table_class(table)
    return table_classes

def generate_table_class(base, table_name):
    return type(table_name, (base,), dict(__tablename__ = table_name,
                                          __table_args__ = {'autoload' : True}))

When I run generate_table_class_list(engine), I got following error messages:

/opt/packages/sqlalchemy/engine/result.pyc in first(self)
    829         try:
    830             if row is not None:
    --> 831                 return self.process_rows([row])[0]
    832             else:
    833                 return None

/opt/packages/sqlalchemy/engine/result.pyc in process_rows(self, rows)
    759         else:
    760             return [process_row(metadata, row, processors, keymap)
    --> 761                     for row in rows]
    762 
    763     def fetchall(self):

TypeError: row must be a sequence

I used similar scripts to do this before with the same version sqlalchemy and it works. However, it doesn't work this time.

Any help is appreciated. Thanks in advance.

Xuan · Accepted Answer · 2014-09-01 04:49:57Z

1

Well after several days researching and testing, I figure it out. When I create an engine, I set the cursor class as SSDictCursor. When I swtiched to SSCursor, it works fine with me, even though I still don't why SSDictCursor breaks my code.

answered Sep 1, 2014 at 4:49

Xuan

414 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

bgth · Accepted Answer · 2017-03-08 06:52:53Z

0

DictCursor is also working. Only thing is you need to specify it when you create the cursor. Please check this one.

connectString = 'mysql+pymysql://{}:{}@{}:{}/{}?charset={}'.format(userDB, passDB, hostDB, portDB, databasename, 'utf8mb4');
engineConnect = create_engine(connectString, connect_args= {'autocommit' : True});

conn = engineConnect.connect();

with conn.connection.cursor(pymysql.cursors.DictCursor) as cursorForFind:
    cursorForFind.execute(queryString, (username));

    (rowCnt, rowVal) = (cursorForFind.rowcount, cursorForFind.fetchone());

answered Mar 8, 2017 at 6:52

bgth

46010 silver badges27 bronze badges

Collectives™ on Stack Overflow

Python SqlAlchemy generate table classes pragmatically

2 Answers 2

Comments

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related