SqlAlchemy learn notes

Readme.md

Cascades

Session Basics

Using Passive Deletes

Contextual/Thread-local Sessions

Managing Transactions

delete-orphan(from stackoverflow)

Object Relational Tutorial

Association Object

Keep going ...

s1.py

# !/usr/bin/env python
# -*- coding:utf8 -*-
'''
Author : myth
Date   : 14-8-11
Email  : belongmyth at 163.com
'''

import sys

reload(sys)
sys.setdefaultencoding(u'utf-8')

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy import Column, ForeignKey, INTEGER, VARCHAR, DATETIME, FLOAT, BOOLEAN, TEXT
from sqlalchemy.orm import relationship, backref

from datetime import datetime, timedelta
import random
import math

# 数据库配置
_db_config = {
    u'db_type': u'mysql',
    u'db_driven': u'mysqldb',
    u'db_user': u'root',
    u'db_passwd': u'bmw12345',
    u'db_host': u'localhost',
    u'db_port': u'3306',
    u'db_selected_db': u'salearn',
    u'db_charset': 'utf8'
}

_db_connect_params = u'%(db_type)s+%(db_driven)s://%(db_user)s:%(db_passwd)s@%(db_host)s/%(db_selected_db)s?charset=%(db_charset)s'

DB_CONNECT_PARAMS = _db_connect_params % _db_config

#DB_CONNECT_PARAMS='mysql+mysqldb://root:root@localhost/account?charset=utf8'
u'''
    :param pool_size=5: the number of connections to keep open
        inside the connection pool. This used with
        :class:`~sqlalchemy.pool.QueuePool` as
        well as :class:`~sqlalchemy.pool.SingletonThreadPool`.  With
        :class:`~sqlalchemy.pool.QueuePool`, a ``pool_size`` setting
        of 0 indicates no limit; to disable pooling, set ``poolclass`` to
        :class:`~sqlalchemy.pool.NullPool` instead.

    :param pool_recycle=-1: this setting causes the pool to recycle
        connections after the given number of seconds has passed. It
        defaults to -1, or no timeout. For example, setting to 3600
        means connections will be recycled after one hour. Note that
        MySQL in particular will disconnect automatically if no
        activity is detected on a connection for eight hours (although
        this is configurable with the MySQLDB connection itself and the
        server configuration as well).
'''
engine = create_engine(DB_CONNECT_PARAMS, echo=True, pool_size=10, pool_recycle=3600)

Session = sessionmaker(bind=engine)
ScopedSession = scoped_session(sessionmaker(bind=engine))


def get_session():
    return ScopedSession()


def get_no_scoped_session():
    return Session()

#DBSession = get_session()

BaseModel = declarative_base(bind=engine)

#BaseModel.objects = DBSession.query_property()  # all model will have it
BaseModel.__table_args__ = {  # 指定所有model的默认参数
                              u'mysql_charset': u'utf8',
                              u'mysql_engine': u'InnoDB'
}


def create_tables():
    BaseModel.metadata.create_all(engine)


def drop_tables():
    BaseModel.metadata.drop_all(engine)


u'''
学生成绩管理系统

班级表

学生表

科目表

成绩表

涉及操作:
  添加/修改/删除班级
  添加/修改/删除学生
  添加/修改/删除科目
  添加/修改/删除学生成绩

'''


class MClass(BaseModel):
    __tablename__ = u'mclass'  #数据库表名称

    id = Column(u'id', INTEGER, primary_key=True)  #班级ID
    name = Column(u'name', VARCHAR(50), unique=True, nullable=False)  #班级名称,唯一且不能为空
    created = Column(u'created', DATETIME, default=datetime.now)

    def __repr__(self):
        return u'<Class %s with %d students, created at %s>' % (
            self.name, len(self.students), self.created.strftime(u'%Y-%m-%d %H:%M:%S'))


class MStudent(BaseModel):
    __tablename__ = u'mstudent'

    id = Column(u'id', INTEGER, primary_key=True)  #学生ID
    class_id = Column(u'class_id', ForeignKey(MClass.id), nullable=False)  #学生所属班级ID
    name = Column(u'name', VARCHAR(30), nullable=False)  #学生名称,必填
    age = Column(u'age', INTEGER)  #学生年龄， 选填
    created = Column(u'created', DATETIME, default=datetime.now)

    cls = relationship(MClass, backref=backref(u'students', cascade=u'all'))  #学生所属班级对象

    def __repr__(self):
        return u'<Student %s in Class %s, created at %s>' % (
            self.name, self.cls.name, self.created.strftime(u'%Y-%m-%d %H:%M:%S'))


class MCourse(BaseModel):
    __tablename__ = u'mcourse'

    id = Column(u'id', INTEGER, primary_key=True)  #课程ID
    name = Column(u'name', VARCHAR(30), unique=True, nullable=False)  #课程名称,唯一且非空
    created = Column(u'created', DATETIME, default=datetime.now)

    def __repr__(self):
        return u'<Course %s, created at %s>' % (self.name, self.created.strftime(u'%Y-%m-%d %H:%M:%S'))


class MScore(BaseModel):
    __tablename__ = u'mscore'

    id = Column(u'id', INTEGER, primary_key=True)  #成绩记录ID
    student_id = Column(u'student_id', ForeignKey(MStudent.id), nullable=False)  #学生ID
    course_id = Column(u'course_id', ForeignKey(MCourse.id), nullable=False)  #课程ID
    score = Column(u'score', FLOAT, default=0.0)

    student = relationship(MStudent, backref=backref(u'scores', cascade=u'all'))
    course = relationship(MCourse, backref=backref(u'scores', cascade=u'all'))

    def __repr__(self):
        return u'<Student %s in Class %s with %.1f for Course %s, created at %s>' % (
            self.student.name, self.student.cls.name, self.score, self.course.name,
            self.created.strftime(u'%Y-%m-%d %H:%M:%S')
        )


cls = MClass
stu = MStudent
c = MCourse
s = MScore


class RandomException(Exception):
    pass


def random_exception():
    if random.random() < 0.0000005:
        raise RandomException()


def random_age():
    return random.randint(20, 30)


def random_score():
    return round(random.random() * 100, 1)


def init_data():
    session = get_session()
    try:
        course_names = [u'语文', u'数学', u'英语']
        courses = []
        # 创建3门课程
        for course_name in course_names:
            course = MCourse()
            course.name = course_name
            random_exception()
            session.add(course)
            courses.append(course)

        # 随机创建3个班级
        for cls_index in range(3):
            # create a class
            cls = MClass()
            cls.name = u'class-%d' % (cls_index + 1, )
            random_exception()

            # 为每个班级随机创建10个学生
            for stu_index in range(10):
                student = MStudent()
                student.name = u'class-%d-student-%d' % (cls_index + 1, stu_index + 1)
                student.age = random_age()
                random_exception()

                # 为每个班级中的学生随机生成每门课程的成绩
                for course in courses:
                    score = MScore()
                    score.score = random_score()
                    score.student = student
                    score.course = course
                    random_exception()
                    #session.add(score)

                #session.add(student)
                cls.students.append(student)

            session.add(cls)

        session.commit()
        return True
    except RandomException:
        print u'random exception happened, rollback all transaction'
        session.rollback()
    finally:
        session.close()
    return False


def random_delete_class():
    session = get_session()
    classes = session.query(MClass).all()
    cls = random.choice(classes)
    print u'will delete class %s with id %d' % (cls.name, cls.id)
    session.delete(cls)
    session.commit()


def random_delete_student():
    session = get_session()
    students = session.query(MStudent).all()
    stu = random.choice(students)
    print u'will delete student %s with id %d' % (stu.name, stu.id)
    session.delete(stu)
    session.commit()


def random_delete_course():
    session = get_session()
    courses = session.query(MCourse).all()
    course = random.choice(courses)
    print u'will delete course %s with id %d' % (course.name, course.id)
    session.delete(course)
    session.commit()


def main():
    # 删除所有的表
    print u'dropping tables ...'
    drop_tables()
    # 创建所有的表
    print u'creating tables ...'
    create_tables()
    # 初始化数据
    print u'init database records ...'
    init_success = init_data()
    if init_success:
        # 随机操作数据
        print u'random delete class ...', u'*' * 50
        random_delete_class()
        print u'random delete student', u'*' * 50
        random_delete_student()
        print u'random delete course', u'*' * 50
        random_delete_course()


if __name__ == u'__main__':
    main()

s2.py

# !/usr/bin/env python
# -*- coding:utf8 -*-
'''
Author : myth
Date   : 14-8-11
Email  : belongmyth at 163.com
'''

import sys

reload(sys)
sys.setdefaultencoding(u'utf-8')

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy import Column, ForeignKey, INTEGER, VARCHAR, DATETIME, FLOAT, BOOLEAN, TEXT
from sqlalchemy.orm import relationship, backref

from datetime import datetime, timedelta
import random
import math

# 数据库配置
_db_config = {
    u'db_type': u'mysql',
    u'db_driven': u'mysqldb',
    u'db_user': u'root',
    u'db_passwd': u'bmw12345',
    u'db_host': u'localhost',
    u'db_port': u'3306',
    u'db_selected_db': u'salearn',
    u'db_charset': 'utf8'
}

_db_connect_params = u'%(db_type)s+%(db_driven)s://%(db_user)s:%(db_passwd)s@%(db_host)s/%(db_selected_db)s?charset=%(db_charset)s'

DB_CONNECT_PARAMS = _db_connect_params % _db_config

#DB_CONNECT_PARAMS='mysql+mysqldb://root:root@localhost/account?charset=utf8'
u'''
    :param pool_size=5: the number of connections to keep open
        inside the connection pool. This used with
        :class:`~sqlalchemy.pool.QueuePool` as
        well as :class:`~sqlalchemy.pool.SingletonThreadPool`.  With
        :class:`~sqlalchemy.pool.QueuePool`, a ``pool_size`` setting
        of 0 indicates no limit; to disable pooling, set ``poolclass`` to
        :class:`~sqlalchemy.pool.NullPool` instead.

    :param pool_recycle=-1: this setting causes the pool to recycle
        connections after the given number of seconds has passed. It
        defaults to -1, or no timeout. For example, setting to 3600
        means connections will be recycled after one hour. Note that
        MySQL in particular will disconnect automatically if no
        activity is detected on a connection for eight hours (although
        this is configurable with the MySQLDB connection itself and the
        server configuration as well).
'''
engine = create_engine(DB_CONNECT_PARAMS, echo=True, pool_size=10, pool_recycle=3600)

Session = sessionmaker(bind=engine)
ScopedSession = scoped_session(sessionmaker(bind=engine))


def get_session():
    return ScopedSession()


def get_no_scoped_session():
    return Session()

#DBSession = get_session()

BaseModel = declarative_base(bind=engine)

#BaseModel.objects = DBSession.query_property()  # all model will have it
BaseModel.__table_args__ = {  # 指定所有model的默认参数
                              u'mysql_charset': u'utf8',
                              u'mysql_engine': u'InnoDB'
}


def create_tables():
    BaseModel.metadata.create_all(engine)


def drop_tables():
    BaseModel.metadata.drop_all(engine)


u'''
学生成绩管理系统

班级表

学生表

科目表

成绩表

涉及操作:
  添加/修改/删除班级
  添加/修改/删除学生
  添加/修改/删除科目
  添加/修改/删除学生成绩

'''


class MClass(BaseModel):
    __tablename__ = u'mclass'  #数据库表名称

    id = Column(u'id', INTEGER, primary_key=True)  #班级ID
    name = Column(u'name', VARCHAR(50), unique=True, nullable=False)  #班级名称,唯一且不能为空
    created = Column(u'created', DATETIME, default=datetime.now)

    students = relationship(u'MStudent', backref=backref(u'cls'), passive_deletes=True)  #学生所属班级对象

    def __repr__(self):
        return u'<Class %s with %d students, created at %s>' % (
            self.name, len(self.students), self.created.strftime(u'%Y-%m-%d %H:%M:%S'))


class MStudent(BaseModel):
    __tablename__ = u'mstudent'

    id = Column(u'id', INTEGER, primary_key=True)  #学生ID
    class_id = Column(u'class_id', ForeignKey(MClass.id, ondelete=u'CASCADE'), nullable=False)  #学生所属班级ID
    name = Column(u'name', VARCHAR(30), nullable=False)  #学生名称,必填
    age = Column(u'age', INTEGER)  #学生年龄， 选填
    created = Column(u'created', DATETIME, default=datetime.now)

    # cls = relationship(u'MClass', backref=backref(u'students'), passive_deletes=True)  #学生所属班级对象
    scores = relationship(u'MScore', backref=backref(u'student'), passive_deletes=True)

    def __repr__(self):
        return u'<Student %s in Class %s, created at %s>' % (
            self.name, self.cls.name, self.created.strftime(u'%Y-%m-%d %H:%M:%S'))


class MCourse(BaseModel):
    __tablename__ = u'mcourse'

    id = Column(u'id', INTEGER, primary_key=True)  #课程ID
    name = Column(u'name', VARCHAR(30), unique=True, nullable=False)  #课程名称,唯一且非空
    created = Column(u'created', DATETIME, default=datetime.now)

    scores = relationship(u'MScore', backref=backref(u'course'), passive_deletes=True)

    def __repr__(self):
        return u'<Course %s, created at %s>' % (self.name, self.created.strftime(u'%Y-%m-%d %H:%M:%S'))


class MScore(BaseModel):
    __tablename__ = u'mscore'

    id = Column(u'id', INTEGER, primary_key=True)  #成绩记录ID
    student_id = Column(u'student_id', ForeignKey(MStudent.id, ondelete=u'CASCADE'), nullable=False)  #学生ID
    course_id = Column(u'course_id', ForeignKey(MCourse.id, ondelete=u'CASCADE'), nullable=False)  #课程ID
    score = Column(u'score', FLOAT, default=0.0)

    # course = relationship(u'MCourse', backref=backref(u'scores'), passive_deletes=True)
    # student = relationship(u'MStudent', backref=backref(u'scores'), passive_deletes=True)

    def __repr__(self):
        return u'<Student %s in Class %s with %.1f for Course %s, created at %s>' % (
            self.student.name, self.student.cls.name, self.score, self.course.name,
            self.created.strftime(u'%Y-%m-%d %H:%M:%S')
        )


cls = MClass
stu = MStudent
c = MCourse
s = MScore


class RandomException(Exception):
    pass


def random_exception():
    if random.random() < 0.0000005:
        raise RandomException()


def random_age():
    return random.randint(20, 30)


def random_score():
    return round(random.random() * 100, 1)


def init_data():
    session = get_session()
    try:
        course_names = [u'语文', u'数学', u'英语']
        courses = []
        # 创建3门课程
        for course_name in course_names:
            course = MCourse()
            course.name = course_name
            random_exception()
            session.add(course)
            courses.append(course)

        # 随机创建3个班级
        for cls_index in range(3):
            # create a class
            cls = MClass()
            cls.name = u'class-%d' % (cls_index + 1, )
            random_exception()

            # 为每个班级随机创建10个学生
            for stu_index in range(10):
                student = MStudent()
                student.name = u'class-%d-student-%d' % (cls_index + 1, stu_index + 1)
                student.age = random_age()
                random_exception()

                # 为每个班级中的学生随机生成每门课程的成绩
                for course in courses:
                    score = MScore()
                    score.score = random_score()
                    score.student = student
                    score.course = course
                    random_exception()
                    #session.add(score)

                #session.add(student)
                cls.students.append(student)

            session.add(cls)

        session.commit()
        return True
    except RandomException:
        print u'random exception happened, rollback all transaction'
        session.rollback()
    finally:
        session.close()
    return False


def random_delete_class():
    session = get_session()
    classes = session.query(MClass).all()
    cls = random.choice(classes)
    print u'will delete class %s with id %d' % (cls.name, cls.id)
    session.delete(cls)
    session.commit()


def random_delete_student():
    session = get_session()
    students = session.query(MStudent).all()
    stu = random.choice(students)
    print u'will delete student %s with id %d' % (stu.name, stu.id)
    session.delete(stu)
    session.commit()


def random_delete_course():
    session = get_session()
    courses = session.query(MCourse).all()
    course = random.choice(courses)
    print u'will delete course %s with id %d' % (course.name, course.id)
    session.delete(course)
    session.commit()


def main():
    # 删除所有的表
    print u'dropping tables ...', u'*' * 50
    drop_tables()
    # 创建所有的表
    print u'creating tables ...', u'*' * 50
    create_tables()
    # 初始化数据
    print u'init database records ...', u'*' * 50
    init_success = init_data()
    if init_success:
        # 随机操作数据
        print u'random delete class ...', u'*' * 50
        random_delete_class()
        print u'random delete student', u'*' * 50
        random_delete_student()
        print u'random delete course', u'*' * 50
        random_delete_course()


if __name__ == u'__main__':
    main()