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

First Post!

Hi, it is my first post in devsniper.com

In devsniper.com i will write about open source technologies, my software development adventures and tutorials.

I hope that it will be useful for me and for all readers -:)