码迷，mamicode.com
首页 > 数据库 > 详细

SQLAchemy

时间：2016-08-06 23:23:45      阅读：267      评论：0      收藏：0      [点我收藏+]

标签：

SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

技术分享

 SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：

MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
  
pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
  
MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
  
cx_Oracle
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
  
更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html

 

一、底层处理

使用 Engine/ConnectionPooling/Dialect 进行数据库操作，Engine使用ConnectionPooling连接数据库，然后再通过Dialect执行SQL语句。

技术分享 
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine
 
 
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
 
# 执行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES (‘1.1.1.22‘, 3)"
# )
 
# 新插入行自增ID
# cur.lastrowid
 
# 执行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES(%s, %s)",[(‘1.1.1.22‘, 3),(‘1.1.1.221‘, 3),]
# )
 
 
# 执行SQL
# cur = engine.execute(
#     "INSERT INTO hosts (host, color_id) VALUES (%(host)s, %(color_id)s)",
#     host=‘1.1.1.99‘, color_id=3
# )
 
# 执行SQL
# cur = engine.execute(‘select * from hosts‘)
# 获取第一行数据
# cur.fetchone()
# 获取第n行数据
# cur.fetchmany(3)
# 获取所有数据
# cur.fetchall()
View Code

二、ORM功能使用

使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有组件对数据进行操作。根据类创建对象，对象转换成SQL，执行SQL。

1、创建表

技术分享 
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)

Base = declarative_base()

# 创建单表
class Users(Base):
    __tablename__ = ‘users‘
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))

    __table_args__ = (
    UniqueConstraint(‘id‘, ‘name‘, name=‘uix_id_name‘),
        Index(‘ix_id_name‘, ‘name‘, ‘extra‘),
    )


# 一对多
class Favor(Base):
    __tablename__ = ‘favor‘
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default=‘red‘, unique=True)


class Person(Base):
    __tablename__ = ‘person‘
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))


# 多对多
class Group(Base):
    __tablename__ = ‘group‘
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)


class Server(Base):
    __tablename__ = ‘server‘

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)


class ServerToGroup(Base):
    __tablename__ = ‘servertogroup‘
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey(‘server.id‘))
    group_id = Column(Integer, ForeignKey(‘group.id‘))


def init_db():
    Base.metadata.create_all(engine)


def drop_db():
    Base.metadata.drop_all(engine)
View Code
技术分享 
#!/usr/bin/env python
# -*- coding:utf-8 -*-


import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey,UniqueConstraint,Index
from sqlalchemy.orm import sessionmaker,relationship


engine = create_engine(‘mysql+pymysql://root:python123@192.168.12.100:3306/test‘)
Base =declarative_base()

class Son(Base):
    __tablename__ = ‘son‘
    id = Column(Integer,primary_key=True)
    name = Column(String(32))
    age = Column(String(16))

    father_id = Column(Integer,ForeignKey(‘father.id‘))

class Father(Base):
    __tablename__ = ‘father‘
    id = Column(Integer,primary_key=True)
    name = Column(String(32))
    age = Column(String(16))
    # son = relationship(‘Son‘)


Base.metadata.create_all(engine) #创建表
#Base.Metadata.drop_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

f1 = Father(name=‘alvin‘,age=50)
session.commit()
w1 = Son(name = ‘little alvin1‘, age = 4, father_id=1)
w2 = Son(name = ‘little alvin1‘, age = 4, father_id=1)
# w1 = Son(name = ‘little alvin1‘, age = 4)
# w2 = Son(name = ‘little alvin1‘, age = 4)

# f1.son=[w1,w2]

session.add_all([f1,w1,w2])
session.commit()
创建表-外键并添加值的方式一
技术分享 
#!/usr/bin/env python
# -*- coding:utf-8 -*-


import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey,UniqueConstraint,Index
from sqlalchemy.orm import sessionmaker,relationship


engine = create_engine(‘mysql+pymysql://root:python123@192.168.12.100:3306/test‘)
Base =declarative_base()

class Son(Base):
    __tablename__ = ‘son‘
    id = Column(Integer,primary_key=True)
    name = Column(String(32))
    age = Column(String(16))

    father_id = Column(Integer,ForeignKey(‘father.id‘))

class Father(Base):
    __tablename__ = ‘father‘
    id = Column(Integer,primary_key=True)
    name = Column(String(32))
    age = Column(String(16))
    son = relationship(‘Son‘)


Base.metadata.create_all(engine) #创建表
#Base.Metadata.drop_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

f1 = Father(name=‘alvin‘,age=50)
session.commit()
# w1 = Son(name = ‘little alvin1‘, age = 4, father_id=1)
# w2 = Son(name = ‘little alvin1‘, age = 4, father_id=1)
w1 = Son(name = ‘little alvin1‘, age = 4)
w2 = Son(name = ‘little alvin1‘, age = 4)

f1.son=[w1,w2]

session.add_all([f1,w1,w2])
session.commit()
创建表-外键并添加值的方式二

2.操作表

技术分享 
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)

Base = declarative_base()

# 创建单表
class Users(Base):
    __tablename__ = ‘users‘
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    extra = Column(String(16))

    __table_args__ = (
    UniqueConstraint(‘id‘, ‘name‘, name=‘uix_id_name‘),
        Index(‘ix_id_name‘, ‘name‘, ‘extra‘),
    )

    def __repr__(self):
        return "%s-%s" %(self.id, self.name)

# 一对多
class Favor(Base):
    __tablename__ = ‘favor‘
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default=‘red‘, unique=True)

    def __repr__(self):
        return "%s-%s" %(self.nid, self.caption)

class Person(Base):
    __tablename__ = ‘person‘
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
    # 与生成表结构无关，仅用于查询方便
    favor = relationship("Favor", backref=‘pers‘)

# 多对多
class ServerToGroup(Base):
    __tablename__ = ‘servertogroup‘
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey(‘server.id‘))
    group_id = Column(Integer, ForeignKey(‘group.id‘))
    group = relationship("Group", backref=‘s2g‘)
    server = relationship("Server", backref=‘s2g‘)

class Group(Base):
    __tablename__ = ‘group‘
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)
    # group = relationship(‘Group‘,secondary=ServerToGroup,backref=‘host_list‘)


class Server(Base):
    __tablename__ = ‘server‘

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)




def init_db():
    Base.metadata.create_all(engine)


def drop_db():
    Base.metadata.drop_all(engine)


Session = sessionmaker(bind=engine)
session = Session()
表结构+连接数据库
技术分享 
obj = Users(name="alex0", extra=‘sb‘)
session.add(obj)
session.add_all([
    Users(name="alex1", extra=‘sb‘),
    Users(name="alex2", extra=‘sb‘),
])
session.commit()
增
技术分享 
session.query(Users).filter(Users.id > 2).delete()
session.commit()
删
技术分享 
session.query(Users).filter(Users.id > 2).update({"name" : "099"})
session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False)
session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate")
session.commit()
改
技术分享 
ret = session.query(Users).all()
ret = session.query(Users.name, Users.extra).all()
ret = session.query(Users).filter_by(name=‘alex‘).all()
ret = session.query(Users).filter_by(name=‘alex‘).first()
查
技术分享 
#　条件
ret = session.query(Users).filter_by(name=‘alex‘).all()
ret = session.query(Users).filter(Users.id > 1, Users.name == ‘eric‘).all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == ‘eric‘).all()
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name=‘eric‘))).all()
from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == ‘eric‘)).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == ‘eric‘)).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == ‘eric‘, Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符
ret = session.query(Users).filter(Users.name.like(‘e%‘)).all()
ret = session.query(Users).filter(~Users.name.like(‘e%‘)).all()

# 限制
ret = session.query(Users)[1:2]

# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()

ret = session.query(Person).join(Favor).all()

ret = session.query(Person).join(Favor, isouter=True).all()


# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()
其他

更多功能参见文档，猛击这里下载PDF

技术分享 
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy import create_engine,and_,or_,func,Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey,UniqueConstraint,DateTime
from  sqlalchemy.orm import sessionmaker,relationship

Base = declarative_base() #生成一个SqlORM 基类

# 服务器账号和组
# HostUser2Group = Table(‘hostuser_2_group‘,Base.metadata,
#     Column(‘hostuser_id‘,ForeignKey(‘host_user.id‘),primary_key=True),
#     Column(‘group_id‘,ForeignKey(‘group.id‘),primary_key=True),
# )

# 用户和组关系表，用户可以属于多个组，一个组可以有多个人
UserProfile2Group = Table(‘userprofile_2_group‘,Base.metadata,
    Column(‘userprofile_id‘,ForeignKey(‘user_profile.id‘),primary_key=True),
    Column(‘group_id‘,ForeignKey(‘group.id‘),primary_key=True),
)

# 程序登陆用户和服务器账户，一个人可以有多个服务器账号，一个服务器账号可以给多个人用
UserProfile2HostUser= Table(‘userprofile_2_hostuser‘,Base.metadata,
    Column(‘userprofile_id‘,ForeignKey(‘user_profile.id‘),primary_key=True),
    Column(‘hostuser_id‘,ForeignKey(‘host_user.id‘),primary_key=True),
)


class Host(Base):
    __tablename__=‘host‘
    id = Column(Integer,primary_key=True,autoincrement=True)
    hostname = Column(String(64),unique=True,nullable=False)
    ip_addr = Column(String(128),unique=True,nullable=False)
    port = Column(Integer,default=22)
    def __repr__(self):
        return  "<id=%s,hostname=%s, ip_addr=%s>" %(self.id,
                                                    self.hostname,
                                                    self.ip_addr)


class HostUser(Base):
    __tablename__ = ‘host_user‘
    id = Column(Integer,primary_key=True)
    AuthTypes = [
        (u‘ssh-passwd‘,u‘SSH/Password‘),
        (u‘ssh-key‘,u‘SSH/KEY‘),
    ]
    # auth_type = Column(ChoiceType(AuthTypes))
    auth_type = Column(String(64))
    username = Column(String(64),unique=True,nullable=False)
    password = Column(String(255))

    host_id = Column(Integer,ForeignKey(‘host.id‘))
    
    # groups = relationship(‘Group‘,
    #                       secondary=HostUser2Group,
    #                       backref=‘host_list‘)

    __table_args__ = (UniqueConstraint(‘host_id‘,‘username‘, name=‘_host_username_uc‘),)

    def __repr__(self):
        return  "<id=%s,name=%s>" %(self.id,self.username)


class Group(Base):
    __tablename__ = ‘group‘
    id = Column(Integer,primary_key=True)
    name = Column(String(64),unique=True,nullable=False)
    def __repr__(self):
        return  "<id=%s,name=%s>" %(self.id,self.name)


class UserProfile(Base):
    __tablename__ = ‘user_profile‘
    id = Column(Integer,primary_key=True)
    username = Column(String(64),unique=True,nullable=False)
    password = Column(String(255),nullable=False)
    # host_list = relationship(‘HostUser‘,
    #                       secondary=UserProfile2HostUser,
    #                       backref=‘userprofiles‘)
    # groups = relationship(‘Group‘,
    #                       secondary=UserProfile2Group,
    #                       backref=‘userprofiles‘)
    def __repr__(self):
        return  "<id=%s,name=%s>" %(self.id,self.username)


class AuditLog(Base):
    __tablename__ = ‘audit_log‘
    id = Column(Integer,primary_key=True)
    userprofile_id = Column(Integer,ForeignKey(‘user_profile.id‘))
    hostuser_id = Column(Integer,ForeignKey(‘host_user.id‘))
    action_choices2 = [
        (u‘cmd‘,u‘CMD‘),
        (u‘login‘,u‘Login‘),
        (u‘logout‘,u‘Logout‘),
    ]
    action_type = Column(ChoiceType(action_choices2))
    #action_type = Column(String(64))
    cmd = Column(String(255))
    date = Column(DateTime)

    # user_profile = relationship("UserProfile")
    #bind_host = relationship("BindHost")


engine = create_engine("mysql+pymsql://root:123@localhost:3306/stupid_jumpserver",echo=False)
Base.metadata.create_all(engine) #创建所有表结构
表结构操作联系

 

---苑昊单表操作

（不涉及一对多，多对多）
技术分享 View Code
技术分享 
#coding:utf8
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker
print(sqlalchemy.__version__)


engine = create_engine(‘sqlite:///dbyuan1.db‘, echo=True)

Base = declarative_base()#生成一个SQLORM基类

class User(Base):
    __tablename__ = ‘users‘

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    password = Column(String)

    def __repr__(self):
       return "<User(name=‘%s‘, fullname=‘%s‘, password=‘%s‘)>" % (
                            self.name, self.fullname, self.password)

Base.metadata.create_all(engine)  #创建所有表结构

ed_user = User(name=‘xiaoyu‘, fullname=‘Xiaoyu Liu‘, password=‘123‘)
print(ed_user)
#这两行触发sessionmaker类下的__call__方法，return得到 Session实例，赋给变量session，所以session可以调用Session类下的add，add_all等方法
MySession = sessionmaker(bind=engine)
session = MySession()

session.add(ed_user)
# our_user = session.query(User).filter_by(name=‘ed‘).first()
# SELECT * FROM users WHERE name="ed" LIMIT 1;
# session.add_all([
#     User(name=‘alex‘, fullname=‘Alex Li‘, password=‘456‘),
#     User(name=‘alex‘, fullname=‘Alex old‘, password=‘789‘),
#     User(name=‘peiqi‘, fullname=‘Peiqi Wu‘, password=‘sxsxsx‘)])

session.commit()

#print(">>>",session.query(User).filter_by(name=‘ed‘).first())
#print(session.query(User).all())
# for row in session.query(User).order_by(User.id):
#      print(row)
# for row in session.query(User).filter(User.name.in_([‘alex‘, ‘wendy‘, ‘jack‘])):＃这里的名字是完全匹配
#     print(row)
# for row in session.query(User).filter(~User.name.in_([‘ed‘, ‘wendy‘, ‘jack‘])):
#     print(row)
#print(session.query(User).filter(User.name == ‘ed‘).count())
#from sqlalchemy import and_, or_

# for row in session.query(User).filter(and_(User.name == ‘ed‘, User.fullname == ‘Ed Jones‘)):
#     print(row)
# for row in session.query(User).filter(or_(User.name == ‘ed‘, User.name == ‘wendy‘)):
#     print(row)
单表操作

 

 二  一对多的关联表操作

案例一

技术分享 
复制代码
#coding:utf8

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey
from sqlalchemy.orm import sessionmaker,relationship


engine = create_engine(‘sqlite:///dbyuan2.db‘, echo=True)

Base = declarative_base()

class Father(Base):
    __tablename__ = ‘father‘
#id = Column(Integer, primary_key=True)里的数据类型一定写整型(Integer)
    id = Column(Integer, primary_key=True)
    name = Column(String(20))
    def __repr__(self):
         return "<Father(name=‘%s‘)>" % self.name


class Son(Base):
    __tablename__ = ‘son‘

    id = Column(Integer, primary_key=True)
    name = Column(String(20))
    #ForeignKey建在多的一方
    father_id = Column(String(20), ForeignKey(‘father.id‘))
    father=relationship("Father",backref="son", order_by=id)
    def __repr__(self):
         return "<Son(name=‘%s‘)>" % self.name

Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()


f1= Father(name=‘zhangsan‘)
f2= Father(name=‘lisi‘)
f3= Father(name=‘wangwu‘)

f1.son = [Son(name=‘zhangdasan‘),Son(name=‘zhangersan‘)]


session.add(f1)
session.commit()

for u, a in session.query(Father, Son).                    filter(Father.id==Son.id).                    all():
    print u, a   #<Father(name=‘zhangsan‘)> <Son(name=‘zhangdasan‘)>
View Code

技术分享

案例二

技术分享 
#__ *__ coding:utf8__*__

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,and_,or_,ForeignKey
from  sqlalchemy.orm import sessionmaker,relationship

Base = declarative_base() #生成一个SqlORM 基类


engine = create_engine(‘sqlite:///dbyuan3.db‘, echo=True)


class Host(Base):
    __tablename__ =‘host‘
    id = Column(Integer,primary_key=True,autoincrement=True)
    hostname = Column(String(64),unique=True,nullable=False)
    ip_addr = Column(String(128),unique=True,nullable=False)
    port = Column(Integer,default=22)
    #前提 一个主机只能属于一个组
    group_id=Column(Integer,ForeignKey(‘group.id‘))
    group=relationship(‘Group‘,backref=‘host‘)
    def __repr__(self):
        return "id:%s hostname:%s port:%s"%(self.id,self.hostname,self.port)


class Group(Base):
    __tablename__=‘group‘
    id=Column(Integer,primary_key=True)
    name=Column(String(64),unique=True,nullable=False)

    def __repr__(self):
        return "id:%s hostname:%s"%(self.id,self.name)


Base.metadata.create_all(engine) #创建所有表结构

if __name__ == ‘__main__‘:
    SessionCls = sessionmaker(bind=engine,autoflush=False)
    session = SessionCls()

    g1=Group(name=‘g1‘)
    g2=Group(name=‘g2‘)
    g3=Group(name=‘g3‘)
    session.add_all([g1,g2,g3])
    session.commit()

    h1 = Host(hostname=‘localhost‘,ip_addr=‘127.0.0.1‘,group_id=g1.id)#g1如果在这之前没有提交，group_id拿到的永远是一个空值
    h2 = Host(hostname=‘ubuntu‘,ip_addr=‘192.168.2.243‘,port=20000)
    session.add_all([h1,h2])

    session.commit()

    g1=session.query(Group).filter(Group.name==‘g1‘).first()
    h=session.query(Host).filter(Host.hostname==‘localhost‘).first()#注意要加上first()，否则报错，注意与all()结果的不同

    print "<<<",g2
    print ">>>",h

    print(h.group.name)
    print g1.host
    print g1.host[0].hostname

    #g2.host什么结果？（未绑定，无结果）
View Code

技术分享

三 多对多的关联表操作

技术分享 
#!/usr/bin/env python
# -*- coding:utf-8 -*-

from sqlalchemy import create_engine,and_,or_,func,Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey
from  sqlalchemy.orm import sessionmaker,relationship

Base = declarative_base() #生成一个SqlORM 基类


Host2Group = Table(‘host_2_group‘,Base.metadata,
    Column(‘host_id‘,ForeignKey(‘host.id‘),primary_key=True),
    Column(‘group_id‘,ForeignKey(‘group.id‘),primary_key=True),)


engine = create_engine(‘sqlite:///dbyuan4.db‘, echo=True)

class Host(Base):
    __tablename__ = ‘host‘

    id = Column(Integer,primary_key=True,autoincrement=True)
    hostname = Column(String(64),unique=True,nullable=False)
    ip_addr = Column(String(128),unique=True,nullable=False)
    port = Column(Integer,default=22)
    group = relationship(‘Group‘,
                          secondary=Host2Group,
                          backref=‘host_list‘)

    #group =relationship("Group",back_populates=‘host_list‘)
    def __repr__(self):
        return  "<id=%s,hostname=%s, ip_addr=%s>" %(self.id,
                                                    self.hostname,
                                                    self.ip_addr)
class Group(Base):
    __tablename__ = ‘group‘
    id = Column(Integer,primary_key=True)
    name = Column(String(64),unique=True,nullable=False)

    def __repr__(self):
        return  "<id=%s,name=%s>" %(self.id,self.name)

Base.metadata.create_all(engine) #创建所有表结构

if __name__ == ‘__main__‘:
    SessionCls = sessionmaker(bind=engine,autoflush=False)
    session = SessionCls()


    g1 = Group(name=‘g1‘)
    g2 = Group(name=‘g2‘)
    g3 = Group(name=‘g3‘)
    g4 = Group(name=‘g4‘)
    session.add_all([g1,g2,g3,g4])
    session.commit()


    #g4 = session.query(Group).filter(Group.name==‘g4‘).first()
    #h = session.query(Host).filter(Host.hostname==‘localhost‘).update({‘group_id‘:g4.id})
    #h = session.query(Host).filter(Host.hostname==‘localhost‘).first()
    #print("h1:",h.group.name )
    #print("g:",g4.host_list )


    h1 = Host(hostname=‘h1‘,ip_addr=‘192.168.1.56‘)
    h2 = Host(hostname=‘h2‘,ip_addr=‘192.168.1.57‘,port=10000)
    h3 = Host(hostname=‘ubuntu‘,ip_addr=‘192.168.1.58‘,port=10000)
    session.add_all([h1,h2,h3])
    session.commit()

    groups = session.query(Group).all()
    g1 = session.query(Group).first()

    h2 = session.query(Host).filter(Host.hostname==‘h2‘).first()
    h2.group = groups[1:-1]
    print("===========>",h2.group)

    #objs = #session.query(Host).join(Host.group).group_by(Group.name).all()
    #objs = session.query(Host,func.count(Group.name)).    #join(Host.group).group_by(Group.name).all()
    #print("-->objs:",objs)
    #print("++>",obj)
    #obj.hostname = "test server"
    #session.delete(obj)
    #objs = session.query(Host).filter(and_(Host.hostname.like("ub%"), Host.port > 20)).all()

    session.commit()
View Code

技术分享

 

注意：

      1   Session = sessionmaker(bind=engine,autoflush=False) 

      2   session.add添加数据到数据后，一定要session.commit()后才能增删改查，否则结果只能为none

      3   session.query(Group).filter(Group.name==‘g1‘).first() 注意有无first()的区别

 

 

再注意：

       1   关于 session.add   session.query   session.commit的顺序问题？

就是说在同一个会话中， insert into table (xxxx)后，可以select * from xxx;可以查询到插入的数据，只是不能在其他会话，比如我另开一个客户端去连接数据库不能查询到刚刚插入的数据。

这个数据已经到数据库。值是数据库吧这个数据给锁了。只有插入数据的那个session可以查看到，其他的session不能查看到，可以理解提交并解锁吧。

      2  第三张表必须利用table创建吗？NO

      3   联合唯一

      4   一对多的第二个例子，如何理解去掉第一个commit后就报错的现象

 

----课上的两个小案例

技术分享 
#!/usr/bin/env python
# -*- coding:utf-8 -*-

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey,UniqueConstraint,Index
from sqlalchemy.orm import sessionmaker,relationship


engine = create_engine(‘mysql+pymysql://root:python123@192.168.12.100:3306/test‘)
Base =declarative_base()

class Son(Base):
    __tablename__ = ‘son‘
    id = Column(Integer,primary_key=True)
    name = Column(String(32))
    age = Column(String(16))

    father_id = Column(Integer,ForeignKey(‘father.id‘))

class Father(Base):
    __tablename__ = ‘father‘
    id = Column(Integer,primary_key=True)
    name = Column(String(32))
    age = Column(String(16))
    son = relationship(‘Son‘,backref = ‘father‘)
    #创建多对多的关系话，得加入那句话：。。。。。

Session = sessionmaker(bind=engine)
session = Session()

ret = session.query(Father.name.label(‘别名‘),Son.name.label("儿子名")).join(Son)

# print(ret)
f1 = session.query(Father).filter_by(id=1).first()
print(f1)
print(f1.son)#这是一个集合【】
print(f1.son[0].name) #想拿到每一个值的话就要for循环
s1 = session.query(Son).filter_by(father_id=1).first()
# print(s1.father.name)
father-son
技术分享 
#!/usr/bin/env python
# -*- coding:utf-8 -*-

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey,UniqueConstraint,Index
from sqlalchemy.orm import sessionmaker,relationship


engine = create_engine(‘mysql+pymysql://root:python123@192.168.12.100:3306/test?charset=utf8‘)
Base =declarative_base()

class Men_to_Wemon(Base):
    __tablename__ = ‘men_to_wemon‘
    nid = Column(Integer, primary_key=True)
    men_id = Column(Integer, ForeignKey(‘men.id‘))
    women_id = Column(Integer, ForeignKey(‘women.id‘))

class Men(Base):
    __tablename__ = ‘men‘
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    age = Column(String(16))
    gf=relationship("Women",secondary=Men_to_Wemon.__table__)

class Women(Base):
    __tablename__ = ‘women‘
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    age = Column(String(16))
    bf = relationship(‘Men‘, secondary=Men_to_Wemon.__table__)
    # bf = relationship(‘Men‘, secondary=Men_to_Wemon.__table__,backref=‘gf‘)
##创建多对多的关系话，得加入那句话：。。。。。sescondary....上面的men
Base.metadata.create_all(engine)
# Base.metadata.drop_all(engine)
"""
1.创建完表
2.插入数据
3.用m1.gf = w1插入数据
"""
Session = sessionmaker(bind=engine)
session = Session()

# w1 = Women(name="alex",age="18")
# w2 = Women(name="wusir",age="18")
# m1 = Men(name="如花",age="40")
# m2 = Men(name = "凤姐",age="40")
# #
# #
# session.add_all([m1,m2,w1,w2,])
# session.commit()
# #
# mw = Men_to_Wemon(men_id=1,women_id=2)
# #
# session.add(mw)
# session.commit()


#-------------------
m1 = session.query(Men).filter_by(id=2).first()#对想
w1 = session.query(Women).all()#对象集合
print(m1.gf)
# m1.gf
# m1.gf是一个列表

# session.add_all([m1,])
# session.commit()
man-women
技术分享 
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine

# engine = create_engine(‘sqlite:///dbyuan674uu.db‘, echo=True)
engine = create_engine(‘mysql+pymysql://root@127.0.0.1:3306/com?charset=utf8‘)  #1 连接已存在的数据库

Base = declarative_base()   #2 创建ORM的基类


class Men_to_Wemon(Base):
    __tablename__ = ‘men_to_wemon‘
    nid = Column(Integer, primary_key=True)
    men_id = Column(Integer, ForeignKey(‘men.id‘))
    women_id = Column(Integer, ForeignKey(‘women.id‘))

class Men(Base):
    __tablename__ = ‘men‘
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    age= Column(String(16))
    gf= relationship("Women", secondary=Men_to_Wemon.__table__)
class Women(Base):
    __tablename__ =‘women‘
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    age= Column(String(16))
    bf=relationship("Men",secondary=Men_to_Wemon.__table__)

Base.metadata.create_all(engine) # 3 在数据库生成表
# 4 通过激活sessionmaker的__call__方法来return一个Session实例（Session类下提供了增删改查的具体方法）
Session = sessionmaker(bind=engine)
session = Session()
# m1=Men(name=‘alex‘,age=18)
# m2=Men(name=‘wusir‘,age=18)
# w1=Women(name=‘如花‘,age=40)
# w2=Women(name=‘铁锤‘,age=45)
# session.add_all([m1,m2,w1,w2,])
# session.commit()
# t1=Men_to_Wemon(men_id=1,women_id=2)


m1=session.query(Men).filter_by(id=2).first()
print(m1)
w1=session.query(Women).all()
m1.gf=w1

session.add_all([m1,])
session.commit()

# 需要注意的地方：
#    1 查询时如果不加all，first等，得到的是sql语句，加上后，才是具体的结果；而all的结果是一个列表。
#    2 m1.gf是一个列表，里面存放着符合条件的对象。
#    3 filter与filter_by的区别：filter是拿键值对的参数，filter_by是拿条件判断的参数。
注释

 

SQLAchemy

标签：

原文地址：http://www.cnblogs.com/renfanzi/p/5744925.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！