Skip to content

Instantly share code, notes, and snippets.

@djrobstep
Created May 18, 2016 14:28
Show Gist options
  • Save djrobstep/998b9779d0bbcddacfef5d76a3d0921a to your computer and use it in GitHub Desktop.
Save djrobstep/998b9779d0bbcddacfef5d76a3d0921a to your computer and use it in GitHub Desktop.
SQLAlchemy ORM code to create SQL tables from a dynamically defined column list
from __future__ import (absolute_import, division, print_function,
unicode_literals)
from sqlalchemy import Column, MetaData, Table, create_engine
from sqlalchemy import String, Integer, Float, BigInteger, DateTime
from sqlalchemy.schema import DropTable, CreateTable
from sqlalchemy.orm import scoped_session, sessionmaker
from contextlib import contextmanager
@contextmanager
def Session(*args, **kwargs):
Session = scoped_session(sessionmaker(
bind=create_engine(*args, **kwargs)))
try:
session = Session()
yield session
session.commit()
except:
session.rollback()
raise
finally:
session.close()
def main():
DB = 'postgresql:///example'
TABLE_SPEC = [
('id', BigInteger),
('name', String),
('t_modified', DateTime),
('whatever', String)
]
TABLE_NAME = 'sample_table'
columns = [Column(n, t) for n, t in TABLE_SPEC]
table = Table(TABLE_NAME, MetaData(), *columns)
with Session(DB, echo=True) as s:
# this is just here to make the script idempotent
s.execute('drop table if exists {}'.format(TABLE_NAME))
table_creation_sql = CreateTable(table)
s.execute(table_creation_sql)
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment