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://[email protected]/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

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

ticketList = getTicketList()

for ticket in ticketList:
    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)


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"))

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()?

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)


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)


2013-10-04 22:36:53,263 INFO sqlalchemy.engine.base.Engine 
CREATE TABLE test_daily_history (
    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)


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.

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__]


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

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


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)