How to dynamically create tables by sqlmodel?

[jaytang0923]

First Check

• [X] I added a very descriptive title to this issue.
• [X] I used the GitHub search to find a similar issue and didn't find it.
• [X] I searched the SQLModel documentation, with the integrated search.
• [X] I already searched in Google "How to X in SQLModel" and didn't find any information.
• [X] I already read and followed all the tutorial in the docs and didn't find an answer.
• [X] I already checked if it is not related to SQLModel but to Pydantic.
• [X] I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• [X] I commit to help with one of those options 👆

Example Code

class DeviceStore(SQLModel):
    """
    Devices of one batch
    """
    # __abstract__ = True
    id: Optional[int] = Field(None, primary_key=True,
                              sa_column_kwargs={"autoincrement": True})
    name: str


class DeviceBatch01(DeviceStore):
    __tablename__ = "devicebatch01"

class DeviceBatch02(DeviceStore):
    __tablename__ = "devicebatch02"

Description

I'm new here and learning to use sqlmodel, it's really great, now I got a question which is how to use sqlmodel dynamically to create tables, all my tables have the same format, just the table names are different, like logtable_2205, logtable_2206, logtable_2207. . . could you guys provide some ideas? thanks a lot.

Operating System

Windows

Operating System Details

No response

SQLModel Version

0.0.6

Python Version

Python 3.8.10

Additional Context

No response

[Niccolum]

@jaytang0923 I think type can help with it. For example DeviceBatch01 = type("DeviceBatch01", (DeviceStore,), {"__tablename__": "devicebatch01"})

But this desire looks like you want something bad.

[phi-friday]

All defined tables(not model, need table=True) are stored in tables({tablename: Table} mapping) of MetaData in the SQLModel class. After you find the table you want, you only need to create that table.

[jaytang0923]

@jaytang0923 I think type can help with it. For example DeviceBatch01 = type("DeviceBatch01", (DeviceStore,), {"__tablename__": "devicebatch01"})

But this desire looks like you want something bad.

thank you @Niccolum , as you suggested, i almost succeeded.the remaining question is how to link foreign keys by type?

cls = type(newtab, (DeviceStore,), {'__tablename__': newtab.lower(),
                                    "tusnid": (int, Field(foreign_key="tusns.id",
                                                          primary_key=True))}, table=True)

take error: column 'tusnid'): Can't generate DDL for NullType(); did you forget to specify a type on this Column?

what i need liks this： tusnid: Optional[int] = Field(None, foreign_key="tusns.id", primary_key=True)

[Niccolum]

@jaytang0923 looks like TypedDict helped with it. Type hinting is impossible with this example of writing code, but typeddict should help out (python 3.8+ in typing or typing-extension on 3.6+)

[Niccolum]

I saw on typeddict and thought, how to use it with your case. Now i have only one vision

from typing import TypedDict, Optional

class MyType(TypedDict):
    __tablename__: str
    tusnid: Optional[int]

table_kwargs: MyType = {'__tablename__': newtab.lower(),
                                    "tusnid": Field(foreign_key="tusns.id", primary_key=True))},

cls = type(newtab, (DeviceStore), table_kwargs)

[LTMullineux]

I had a similar problem with trying to create a factory function for abstract columns. I think the below should help solve the issue.

I added some tests below too that prints out the SQL during the table creation, you can see from that the tables/columns/primary key/foreign key creation.

import asyncio
from typing import Optional

import pytest
import pytest_asyncio
from sqlalchemy import Column, ForeignKey, Integer
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel import Field, SQLModel


class Tusns(SQLModel, table=True):
    __tablename__ = "tusns"

    id: int = Field(primary_key=True, nullable=False)


def device_batch_table_factory(suffix: str) -> SQLModel:
    class DeviceBatchBase(SQLModel, table=True):
        __tablename__ = f"device_batch_{suffix}"
        name: str
        id: Optional[int] = Field(
            None,
            sa_column=Column(
                Integer,
                ForeignKey("tusns.id", ondelete="CASCADE"),
                primary_key=True,
                autoincrement=True,
                default=None,
                nullable=True,
            ),
        )

    return DeviceBatchBase


tbl1 = device_batch_table_factory("01")
tbl2 = device_batch_table_factory("02")
tbl3 = device_batch_table_factory("03")


# Tests
@pytest.fixture(scope="session")
def event_loop(request):
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="class")
async def engine(event_loop):
    engine = create_async_engine(
        "postgresql+asyncpg://postgres:postgres@postgres-test:5432/postgres",
        echo=True,
        future=True,
    )

    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)
        await conn.run_sync(SQLModel.metadata.create_all)

    yield engine

    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)

    engine.sync_engine.dispose()


@pytest_asyncio.fixture()
async def session(engine):
    SessionLocal = sessionmaker(
        bind=engine,
        class_=AsyncSession,
        expire_on_commit=False,
        autocommit=False,
        autoflush=False,
    )

    async with engine.connect() as conn:
        tsx = await conn.begin()
        async with SessionLocal(bind=conn) as session:
            nested_tsx = await conn.begin_nested()
            yield session

            if nested_tsx.is_active:
                await nested_tsx.rollback()
            await tsx.rollback()


@pytest.mark.asyncio
class TestTables:
    async def test_foo(self, session: AsyncSession):
        ...

Running this script with pytest spits out the following DDL:

2023-07-28 16:28:00,827 INFO sqlalchemy.engine.Engine
CREATE TABLE tusns (
        id SERIAL NOT NULL,
        PRIMARY KEY (id)
)


2023-07-28 16:28:00,827 INFO sqlalchemy.engine.Engine [no key 0.00015s] ()
2023-07-28 16:28:00,836 INFO sqlalchemy.engine.Engine
CREATE TABLE device_batch_01 (
        id SERIAL,
        name VARCHAR NOT NULL,
        PRIMARY KEY (id),
        FOREIGN KEY(id) REFERENCES tusns (id) ON DELETE CASCADE
)


2023-07-28 16:28:00,836 INFO sqlalchemy.engine.Engine [no key 0.00024s] ()
2023-07-28 16:28:00,846 INFO sqlalchemy.engine.Engine
CREATE TABLE device_batch_02 (
        id SERIAL,
        name VARCHAR NOT NULL,
        PRIMARY KEY (id),
        FOREIGN KEY(id) REFERENCES tusns (id) ON DELETE CASCADE
)


2023-07-28 16:28:00,847 INFO sqlalchemy.engine.Engine [no key 0.00023s] ()
2023-07-28 16:28:00,857 INFO sqlalchemy.engine.Engine
CREATE TABLE device_batch_03 (
        id SERIAL,
        name VARCHAR NOT NULL,
        PRIMARY KEY (id),
        FOREIGN KEY(id) REFERENCES tusns (id) ON DELETE CASCADE
)
...