How to create only one table with SQLAlchemy?

Question

I am unable to create a single table using SQLAlchemy.

I can create it by calling Base.metadata.create_all(engine) but as the number of table grows, this call takes a long time.

I create table classes on the fly and then populate them.

from sqlalchemy import create_engine, Column, Integer, Sequence, String, Date, Float, BIGINT
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class HistoricDay():

    id = Column(Integer, Sequence('id_seq'), primary_key=True)
    #  Date, Open, High, Low, Close, Volume, Adj Close
    date = Column(Date)
    open = Column(Float)
    high = Column(Float)
    low = Column(Float)
    close = Column(Float)
    volume = Column(BIGINT)
    adjClose = Column(Float)

    def __init__(self, date, open, high, low, close, volume, adjClose):
        self.date = date
        self.open = open
        self.high = high
        self.low = low
        self.close = close
        self.volume = volume
        self.adjClose = adjClose

def build_daily_history_table_repr(self):
        return "<"+self.__tablename__+"('{}','{}','{}','{}','{}','{}','{}','{}')>".format(self.id, self.date, self.open, self.high, self.low, self.close, self.volume, self.adjClose)

def build_daily_history_table(ticket):
    classname = ticket+"_HistoricDay"
    globals()[classname] = type(classname, (HistoricDay,Base), {'__tablename__' : ticket+"_daily_history"})
    setattr(globals()[classname], '__repr__', build_daily_history_table_repr)

# Initialize the database :: Connection & Metadata retrieval
engine = create_engine('mysql+cymysql://root@localhost/gwc?charset=utf8&use_unicode=0', pool_recycle=3600) # ,echo = True

# SqlAlchemy :: Session setup
Session = sessionmaker(bind=engine)

# Create all tables that do not already exist
Base.metadata.create_all(engine)

# SqlAlchemy :: Starts a session
session = Session()

ticketList = getTicketList()

for ticket in ticketList:
    build_daily_history_table(ticket)
    class_name = ticket+"_HistoricDay"

    meta_create_all_timer = time.time()
    # Create all tables that do not already exist
    # globals()[class_name]('2005-07-24',0,0,0,0,0,0).create(engine)  #doesn't work
    #(globals()[class_name]).__table__.create(engine) #doesn't work
    # session.commit() #doesn't work

    #Base.metadata.create_all(engine) # works but gets very slow
    print("  meta_create_all_timer {}s".format(time.time()-meta_create_all_timer))

    data = getData(ticket)

    for m_date, m_open, m_close, m_high, m_low, m_volume, m_adjClose in data:
        entry = globals()[class_name](m_date, m_open, m_high, m_low, m_close, m_volume, m_adjClose)
        session.add(entry)

    session.commit()

I saw in the documentation that you can do

engine = create_engine('sqlite:///:memory:')

meta = MetaData()

employees = Table('employees', meta,
    Column('employee_id', Integer, primary_key=True),
    Column('employee_name', String(60), nullable=False, key='name'),
    Column('employee_dept', Integer, ForeignKey("departments.department_id"))
)
employees.create(engine)

However, I'm not able to figure out how to do the same thing as Table does, with declarative_base().

How can I do that with classes that inherit from declarative_base()?

Answer 1

A bit late to the party, but I found creating tables dynamically (have them vary by name and columns) to be painful with SQLAlchemy ORM. I realized it was much easier for me to just write an SQL query and then run the text directly. So, not very neat, but useful so I'll leave it here. It's all VARCHAR(255) since I do transformations later in the pipeline.

def table_creator(table_name: str, selectors: list) -> str:
    columns = [
        "id INTEGER PRIMARY KEY AUTO_INCREMENT",
        ...
    ]

    for selector in selectors:
        columns.append(f"`{selector.name}` VARCHAR(255) {'NOT NULL' if selector.required else ''}")

    query = f"CREATE TABLE {table_name} ({', '.join(columns)})"
    print(query)
    return query

After this you can just run (if you are running async):

from sqlalchemy import text
statement = table_creator(table_name=name, selectors=selectors)
async with CORE_DATABASE.get_session() as sess:
    """
    CORE_DATABASE.get_session() comes down to:
    def session(self) -> AsyncSession:
        return AsyncSession(bind=self._engine)
    """
    await sess.execute(text(statement))

This will create the table you need.

However, as a consequence, you will not have table ORM object to work with immediately.

Answer 2

To create specific tables, giving tables parameter to create_all() method is enough.

Base.metadata.create_all(engine, tables=table_objects)

table_objects equals to:

table_objects = [HistoricDay.__table__]

or

table_objects = [Base.metadata.tables["historicday"]]

I showed one table here. You can increase the number of the tables as you wish.

Reference: http://docs.sqlalchemy.org/en/latest/core/metadata.html#sqlalchemy.schema.MetaData.create_all

Answer 3

Create all not exists tables with one line. It will check whether the table exists first by default.

Base.metadata.create_all(db_engine, Base.metadata.tables.values(),checkfirst=True)

Create one target table with table_name.

Base.metadata.create_all(db_engine, Base.metadata.tables[table_name],checkfirst=True)

It works perfectly with declarative_base.

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
TABLE_PREFIX = "decision_"


class Stock(Base):
    __tablename__ = '{}stocks'.format(TABLE_PREFIX)

    id = Column(Integer, primary_key=True)
    name = Column(String)

class StagePerformance(Base):
    __tablename__ = '{}stage_performance'.format(TABLE_PREFIX)

    id = Column(Integer, primary_key=True)
    date = Column(DateTime)
    stock = relationship("Stock", back_populates="stage_performances")
    period = Column(Integer )
    open = Column(Float)
    high = Column(Float)
    low = Column(Float)
    close = Column(Float)
    change_ratio = Column(Float)
    turnover = Column(Float)
    volume = Column(Float)

Answer 4

Above, the declarative_base() callable returns a new base class from which all mapped classes should inherit. When the class definition is completed, a new Table and mapper() will have been generated.

The resulting table and mapper are accessible via __table__ and __mapper__ attributes

(From here)

Therefore:

def build_daily_history_table(ticket):
    classname = ticket + "_HistoricDay"
    ticket = type(classname, (Base, HistoricDay), {'__tablename__' : ticket+"_daily_history"})
    ticket.__repr__ =  build_daily_history_table_repr
    return ticket

build_daily_history_table("test").__table__.create(bind = engine)

Output:

2013-10-04 22:36:53,263 INFO sqlalchemy.engine.base.Engine 
CREATE TABLE test_daily_history (
    id INTEGER NOT NULL, 
    date DATE, 
    open FLOAT, 
    high FLOAT, 
    low FLOAT, 
    close FLOAT, 
    volume BIGINT, 
    "adjClose" FLOAT, 
    PRIMARY KEY (id)
)


2013-10-04 22:36:53,263 INFO sqlalchemy.engine.base.Engine ()
2013-10-04 22:36:53,263 INFO sqlalchemy.engine.base.Engine COMMIT

Credit goes to javex's comment/correction or I might have suggested something akin to:

Base.metadata.tables["ticket_daily_history"].create(bind = engine)

Advise:

The approach used in build_daily_history_table could be one of the least elegant ways of doing things, primarily for the reason that it is polluting/cluttering the namespace.