In my program I have several classes defined which may or may not be instantiated in a given run time. Therefore I don't want to define the tables in the class and have them all populated together using meta.create_all
leaving my database with a bunch of empty tables. Rather when an instance of a class is created I want it to create the table manually. I have it all working except for one step. The code below is working but it tries to recreate all tables. When I try to use meta.create_all
with a list containing the table I want to create (see commented line) then it doesn't work and I get the following error. How do I get this to work?
AttributeError: 'AsyncEngine' object has no attribute '_run_ddl_visitor'
import asyncio
import logging
from sqlalchemy import MetaData, Table, Column, Integer, String
from sqlalchemy.ext.asyncio import create_async_engine
from functools import partial
import logging
logger = logging.getLogger()
class DB(): pass
db = DB()
class Reg():
def __init__(self, name) -> None:
self.name = name
self.table = self.getTable()
def getTable(self) -> Table:
return Table(
'items',
db.meta,
Column('id', Integer, primary_key=True),
Column('name', String(30)),
Column('type', String(15))
)
async def createTable(self, table:Table):
async with db.engine.begin() as conn:
await conn.run_sync(db.meta.create_all)
# await conn.run_sync(partial(db.meta.create_all,db.engine,[table])) # <--- this gives me the error
async def main():
logging.basicConfig(level=logging.INFO)
db.engine = create_async_engine(f'sqlite+aiosqlite:///:memory:', echo=True)
db.meta = MetaData()
reg = Reg('Reg')
await reg.createTable(reg.table)
await db.engine.dispose()
if __name__ == '__main__':
asyncio.run(main())
python 3.12
sqlalchemy 2.0.35
While you can't use a partial
here, you can use a lambda
:
await conn.run_sync(lambda conn: db.meta.create_all(conn, [table]))