python-3.xsqlalchemypython-asyncio

How to create a single table at run time with SQLAlchemy asyncio


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

Solution

  • 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]))