创建外键
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Integer
from sqlalchemy import ForeignKey
Base = declarative_base() #生成orm基类
class User(Base):
__tablename__ = 'user'
id = Column(Integer, autoincrement=True, primary_key=True)
name = Column(String(20), nullable=False)
class Addresss(Base):
__tablename__ = 'addresss'
id = Column(Integer, autoincrement=True, primary_key=True)
email_address = Column(String(32), nullable=False)
user_info = Column(Integer, ForeignKey('user.id'))
#user = relationship(User, backref='addresss')
engine = create_engine('mysql+pymysql://alex:123456@192.168.181.128:3306/db_again3',encoding='utf-8')
Base.metadata.create_all(engine)
DBSession = sessionmaker(bind=engine)
session = DBSession()
session.add_all([User(name='alex',),
User(name='jack'),
User(name='jim')])
session.add_all([Addresss(email_address='11.com',user_info=1),
Addresss(email_address='22.com',user_info=2)])
session.commit()
session.close()
关联查询
from sqlalchemy import Column, String, create_engine
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Integer
from sqlalchemy import ForeignKey
Base = declarative_base() #生成orm基类
class User(Base):
__tablename__ = 'user'
id = Column(Integer, autoincrement=True, primary_key=True)
name = Column(String(20), nullable=False)
def __repr__(self):
return "id;%s name:%s" %(self.id, self.name)
class Addresss(Base):
__tablename__ = 'addresss'
id = Column(Integer, autoincrement=True, primary_key=True)
email_address = Column(String(32), nullable=False)
user_info = Column(Integer, ForeignKey('user.id'))
user = relationship(User, backref='addresss') #使两个对象在内存中可以互相调用
def __repr__(self):
return "name:%s email_address:%s user_info:%s" %(self.user.name, self.email_address, self.user_info)
engine = create_engine('mysql+pymysql://alex:123456@192.168.181.128:3306/db_again3',encoding='utf-8')
#Base.metadata.create_all(engine)
DBSession = sessionmaker(bind=engine)
session = DBSession()
"""
session.add_all([User(name='alex',),
User(name='jack'),
User(name='jim')])
session.add_all([Addresss(email_address='11.com',user_info=1),
Addresss(email_address='22.com',user_info=2)])
"""
user_obj = session.query(User).filter(User.name=='jack').first()
print(user_obj.id, user_obj.name)
print(user_obj)
print('===============')
print(user_obj.addresss)
for i in user_obj.addresss: #通过user对象反查关联的address记录
print(i)
print('===============')
address_obj = session.query(Addresss).filter(Addresss.email_address=='11.com').first()
print(address_obj.user) #在address对象里直接查关联的user记录
session.commit()
session.close()
执行结果：
