SQLAlchemy Tutorial 1 : Declarative MySQL Tables

SQLAlchemy Tutorial 1 : Declarative MySQL Tables

SQLAlchemy is awesome ORM and SQL Toolkit for Python. I am using it for MySQL 5.1 with MySQL specific settings.

How can i define and create MySQL tables with SQLAlchemy ORM?

I have 3 Tables :

--
-- customer db schema
-- version 1.0
--

SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0;
SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0;
SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='TRADITIONAL';

DROP SCHEMA IF EXISTS customerdb;
CREATE SCHEMA customerdb;
USE customerdb;

--
-- Table structure for table country
--
DROP TABLE IF EXISTS country;

CREATE TABLE country (
	country_id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
	code VARCHAR(10) NOT NULL,
	name VARCHAR(50) NOT NULL,
	created_by_user VARCHAR(50) DEFAULT NULL,
	created_at DATETIME NOT NULL,
	updated_by_user VARCHAR(50) DEFAULT NULL,
	updated_at DATETIME DEFAULT NULL,
	PRIMARY KEY (country_id),
	UNIQUE KEY (code),
	UNIQUE KEY (name)
)ENGINE=InnoDB DEFAULT CHARSET=utf8;

--
-- Table structure for table customer category
--
DROP TABLE IF EXISTS category;

CREATE TABLE category (
	category_id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
	name VARCHAR(50) NOT NULL,
	created_by_user VARCHAR(50) DEFAULT NULL,
	created_at DATETIME NOT NULL,
	updated_by_user VARCHAR(50) DEFAULT NULL,
	updated_at DATETIME DEFAULT NULL,
	PRIMARY KEY (category_id),
	UNIQUE KEY (name)
)ENGINE=InnoDB DEFAULT CHARSET=utf8;

--
-- Table structure for table customers
--
DROP TABLE IF EXISTS customer;

CREATE TABLE customer (
	customer_id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
	company_name VARCHAR(100) NOT NULL,
	category_id BIGINT UNSIGNED NOT NULL,
	contact_title VARCHAR(50) DEFAULT NULL,
	contact_first_name VARCHAR(50) DEFAULT NULL,
	contact_last_name VARCHAR(50) DEFAULT NULL,
	address TEXT DEFAULT NULL,
	city VARCHAR(50) DEFAULT NULL,
	region VARCHAR(50) DEFAULT NULL,
	postal_code VARCHAR(50) DEFAULT NULL,
	country_id BIGINT UNSIGNED DEFAULT NULL,
	phone VARCHAR(50) DEFAULT NULL,
	fax VARCHAR(50) DEFAULT NULL,
	mobile VARCHAR(50) DEFAULT NULL,
	email VARCHAR(50) DEFAULT NULL,
	homepage VARCHAR(50) DEFAULT NULL,
	skype VARCHAR(50) DEFAULT NULL,
	active BOOLEAN NOT NULL DEFAULT TRUE,
	notes TEXT DEFAULT NULL,
	created_by_user VARCHAR(50) DEFAULT NULL,
	created_at DATETIME NOT NULL,
	updated_by_user VARCHAR(50) DEFAULT NULL,
	updated_at DATETIME DEFAULT NULL,
	PRIMARY KEY (customer_id),
	UNIQUE KEY (company_name),
	KEY idx_fk_category_id (category_id),
	CONSTRAINT fk_customer_category FOREIGN KEY (category_id) REFERENCES category (category_id) ON DELETE RESTRICT ON UPDATE CASCADE,
	KEY idx_fk_country_id (country_id),
	CONSTRAINT fk_customer_country FOREIGN KEY (country_id) REFERENCES country (country_id) ON DELETE RESTRICT ON UPDATE CASCADE
)ENGINE=InnoDB DEFAULT CHARSET=utf8;

SET SQL_MODE=@OLD_SQL_MODE;
SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS;
SET UNIQUE_CHECKS=@OLD_UNIQUE_CHECKS;

and SQLAlchemy Classes :

"""
db entites, mappers and db table settings
"""
from datetime import datetime
from sqlalchemy import Column, ForeignKey, create_engine
from sqlalchemy.dialects.mysql import BIGINT, DATETIME, TEXT, TINYINT, VARCHAR
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm.interfaces import MapperExtension

Base = declarative_base()

class Country(Base):
    """ Country entity class """
    __tablename__ = 'country'
    __table_args__ = {
        'mysql_engine': 'InnoDB',
        'mysql_charset': 'utf8'
    } 

    # this field is in model "id", in db "country_id"
    # not null and auto_increment settings with "primary_key"
    id = Column('country_id', BIGINT(unsigned=True), primary_key=True)
    code = Column(VARCHAR(10), nullable=False, unique=True)
    name = Column(VARCHAR(100), nullable=False, unique=True)
    created_by_user = Column(VARCHAR(50))
    created_at = Column(DATETIME)
    updated_by_user = Column(VARCHAR(50))
    updated_at = Column(DATETIME)

    def __init__(self, code, name):
        self.code = code
        self.name = name

class Category(Base):
    """ Category entity class """
    __tablename__ = 'category'
    __table_args__ = {
        'mysql_engine': 'InnoDB',
        'mysql_charset': 'utf8'
    } 

    id = Column('category_id', BIGINT(unsigned=True), primary_key=True)
    name = Column(VARCHAR(100), nullable=False, unique=True)
    created_by_user = Column(VARCHAR(50))
    created_at = Column(DATETIME)
    updated_by_user = Column(VARCHAR(50))
    updated_at = Column(DATETIME)

    def __init__(self, name):
        self.name = name

class Customer(Base):
    """ Customer entity class """
    __tablename__ = 'customer'
    __table_args__ = {
        'mysql_engine': 'InnoDB',
        'mysql_charset': 'utf8'
    } 

    id = Column('customer_id', BIGINT(unsigned=True), primary_key=True)
    company_name = Column(VARCHAR(100), nullable=False, unique=True)
    # declarative foreing key setting
    # nullable = false, the customer must have a category
    category_id = Column(BIGINT(unsigned=True), ForeignKey('category.category_id',
        name="fk_customer_category", onupdate='CASCADE', ondelete='RESTRICT'), nullable=False)
    contact_title = Column(VARCHAR(50))
    contact_first_name = Column(VARCHAR(50))
    contact_last_name = Column(VARCHAR(50))
    address = Column(TEXT)
    city = Column(VARCHAR(50))
    region = Column(VARCHAR(50))
    postal_code = Column(VARCHAR(50))
	# declarative foreing key setting
    country_id = Column(BIGINT(unsigned=True), ForeignKey('country.country_id',
        name='fk_customer_country', onupdate='CASCADE', ondelete='RESTRICT'))
    phone = Column(VARCHAR(50))
    fax = Column(VARCHAR(50))
    mobile = Column(VARCHAR(50))
    email = Column(VARCHAR(50))
    homepage = Column(VARCHAR(50))
    skype = Column(VARCHAR(50))
	# this customer active or inactive?
    active = Column(TINYINT(1), nullable=False, default=1)
    notes = Column(TEXT)
    created_by_user = Column(VARCHAR(50))
    created_at = Column(DATETIME)
    updated_by_user = Column(VARCHAR(50))
    updated_at = Column(DATETIME)

    def __init__(self, company_name, category_id):
        self.company_name = company_name
        self.category_id = category_id

11 Comments

  1. anoop
    Apr 09, 2012 @ 18:43:32

    i tried the same . but I am getting error again and again while trying to import the module.

    Traceback (most recent call last):
    File “”, line 1, in
    File “build/bdist.linux-x86_64/egg/sqlalchemy/ext/declarative.py”, line 1342, in __init__
    File “build/bdist.linux-x86_64/egg/sqlalchemy/ext/declarative.py”, line 1246, in _as_declarative
    AttributeError: ‘str’ object has no attribute ‘c’

    Base = declarative_base()
    #===============================================================================
    # CountryORM
    #===============================================================================
    class CountryORM(Base):
    ”’
    classdocs
    ”’
    __table__ = ‘country’
    __table_args__ = {
    ‘mysql_engine’: ‘InnoDB’,
    ‘mysql_charset’: ‘utf8′
    }
    id = Column( INTEGER(11), primary_key = True)
    code = Column( VARCHAR(8), nullable = False, unique = True)
    country = Column( VARCHAR(32), nullable = True)
    currency = Column( CHAR(3), nullable = True)

    #===========================================================================
    # __init__
    #===========================================================================
    def __init__(self, code):

    self.code = code

    Reply

    • Cem Ikta
      Apr 10, 2012 @ 16:04:58

      Hi,

      in your code __table__ = ‘country’
      in my code __tablename__ = ‘country’

      I hope it helps.

      Reply

      • anoop
        Apr 10, 2012 @ 16:15:08

        thanks. actually i figured it out after posting the query only. thanks a lot

        Reply

      • zhaogp
        Dec 26, 2016 @ 11:55:35

        thank you very much

        Reply

  2. Tyler Long
    Apr 27, 2012 @ 12:01:01

    Kind of complex. Try quick_orm: https://github.com/tylerlong/quick_orm. You don’t have to write any SQL, and you can build your models in minutes.

    Reply

    • Cem Ikta
      Apr 28, 2012 @ 12:28:33

      For MySQL Tables you do not need to write SQL. SQLAlchemy automatically created for you MySQL Tables from your models ! I’ve written SQL, because I just wanted to show the MySQL database structure. Incidentally, SQLAlchemy is stable, robust and very well.

      Reply

  3. jack
    Mar 22, 2013 @ 17:46:42

    It’s possible to declare each class in a separated file?

    Reply

    • Cem Ikta
      Mar 22, 2013 @ 17:56:37

      Hi,

      Yes it is possible. If you don’t have so many tables, all in one is useful.

      Reply

  4. CD
    Apr 02, 2015 @ 08:50:06

    Where did you defined “EntityMixin”?
    What it mean here ?

    Reply

    • Cem Ikta
      Apr 02, 2015 @ 13:14:27

      Sorry, It was a typo. It is actually BaseEntity, I have corrected in the tutorial. Please see SQLAlchemy Tutorial 2 and 3 for BaseExtension and BaseEntity.

      Reply

  5. Emmanuel
    Apr 13, 2015 @ 02:24:00

    Thank you alot amigo =)

    Reply

Leave a Reply