Python全栈（七）Flask框架之7.ORM过滤条件、外键约束和表关系

Petunia ·

更新时间:2024-11-13

· 599 次阅读

ORM过滤条件、外键约束和表关系一、Flask数据库过滤条件二、ORM模型的外键约束1.ORM建立外键关系2.ORM数据库外键约束3.存在外键时数据的查询三、Flask数据库表关系1.一对多关系 一、Flask数据库过滤条件

过滤条件一般是通过filter()方法实现的，常见的过滤条件如下：

equals

filter(Book.name == 'Name 3')

not equals

filter(Book.name != 'Name 3')

filter(Book.name.like("%Name%"))

filter(Book.name.in_(['Name 1', 'Name 2']))
# in_的参数也可以是query的查询结果
filter(Book.name.in_(session.query(Book.name).filter(Book.name.like('%Name%'))))

not in

filter(~Book.name.in_(['Name 1', 'Name 2']))

is null

filter(Book.name == None)
# 或者是
filter(Book.name.is_(None))

is not null

filter(User.name != None)
# 或者是
filter(User.name.isnot(None))

and

filter(Book.name == 'Name 1', Book.price <= 50).all()
filter(and_(Book.name == 'Name 1', Book.price <= 50)).all()
filter(Book.name == 'Name 1').filter(Book.price <= 50).all()

filter(or_(Book.name == 'Name 1', Book.price >= 50)).all()

equals测试如下：

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy import create_engine, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_demo'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    price = Column(Float)
    def __str__(self):
        return 'Book(id:{}, name:{}， price：{})'.format(self.id, self.name, self.price)
Session = sessionmaker(bind=engine)
session = Session()
results = session.query(Book).filter(Book.name == 'Name 3').all()
for result in results:
    print(result)

打印：

Book(id:3, name:Name 3， price：14.0)

not equals测试：

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy import create_engine, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_demo'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    price = Column(Float)
    def __str__(self):
        return 'Book(id:{}, name:{}， price：{})'.format(self.id, self.name, self.price)
Session = sessionmaker(bind=engine)
session = Session()
results = session.query(Book).filter(Book.name != 'Name 3').all()
for result in results:
    print(result)

打印：

Book(id:1, name:Name 1， price：45.0)
Book(id:2, name:Name 2， price：49.0)
Book(id:4, name:Name 4， price：16.0)
Book(id:5, name:Name 5， price：17.0)
Book(id:6, name:Name 6， price：3.0)
Book(id:7, name:Name 7， price：25.0)
Book(id:8, name:Name 8， price：16.0)
Book(id:9, name:Name 9， price：32.0)
Book(id:10, name:Name 10， price：22.0)

like一般用于模糊查询，测试如下：

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy import create_engine, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_demo'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    price = Column(Float)
    def __str__(self):
        return 'Book(id:{}, name:{}， price：{})'.format(self.id, self.name, self.price)
Session = sessionmaker(bind=engine)
session = Session()
results = session.query(Book).filter(Book.name.like('%me%')).all()
for result in results:
    print(result)

打印：

Book(id:1, name:Name 1， price：45.0)
Book(id:2, name:Name 2， price：49.0)
Book(id:3, name:Name 3， price：14.0)
Book(id:4, name:Name 4， price：16.0)
Book(id:5, name:Name 5， price：17.0)
Book(id:6, name:Name 6， price：3.0)
Book(id:7, name:Name 7， price：25.0)
Book(id:8, name:Name 8， price：16.0)
Book(id:9, name:Name 9， price：32.0)
Book(id:10, name:Name 10， price：22.0)

其中，%表示匹配表示零个或者多个任意字符，只要name字段中含有me字符串，就会被查询到；
还可以使用下划线_，表示一个任意字符（必须有一个字符）。

in测试：

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy import create_engine, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_demo'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    price = Column(Float)
    def __str__(self):
        return 'Book(id:{}, name:{}， price：{})'.format(self.id, self.name, self.price)
Session = sessionmaker(bind=engine)
session = Session()
results = session.query(Book).filter(Book.name.in_(['Name 2', 'Name 5'])).all()
for result in results:
    print(result)

打印：

Book(id:2, name:Name 2， price：49.0)
Book(id:5, name:Name 5， price：17.0)

为了避免和Python中的关键字in冲突，ORM将in后面添加一个下划线；
除了列表，in_()中也可以传入元组，但是建议使用列表。

not in两种方式测试：

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy import create_engine, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_demo'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50), nullable=False)
    price = Column(Float)
    def __str__(self):
        return 'Book(id:{}, name:{}， price：{})'.format(self.id, self.name, self.price)
Session = sessionmaker(bind=engine)
session = Session()
results1 = session.query(Book).filter(~Book.name.in_(['Name 2', 'Name 5'])).all()
print('~ test:')
for result in results1:
    print(result)
results2 = session.query(Book).filter(Book.name.notin_(['Name 2', 'Name 5'])).all()
print('Notin test:')
for result in results2:
    print(result)

打印：

~ test:
Book(id:1, name:Name 1， price：45.0)
Book(id:3, name:Name 3， price：14.0)
Book(id:4, name:Name 4， price：16.0)
Book(id:6, name:Name 6， price：3.0)
Book(id:7, name:Name 7， price：25.0)
Book(id:8, name:Name 8， price：16.0)
Book(id:9, name:Name 9， price：32.0)
Book(id:10, name:Name 10， price：22.0)
Notin test:
Book(id:1, name:Name 1， price：45.0)
Book(id:3, name:Name 3， price：14.0)
Book(id:4, name:Name 4， price：16.0)
Book(id:6, name:Name 6， price：3.0)
Book(id:7, name:Name 7， price：25.0)
Book(id:8, name:Name 8， price：16.0)
Book(id:9, name:Name 9， price：32.0)
Book(id:10, name:Name 10， price：22.0)

显然，not in两种实现方式（ ~ 取反和notin_()方法）的查询结果是一样的，但是使用notin_()方法的可读性更高。

null和not null测试：
修改book表name字段允许为null之后手动插入一条name字段为空的数据，再测试：

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy import create_engine, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_demo'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50))
    price = Column(Float)
    def __str__(self):
        return 'Book(id:{}, name:{}， price：{})'.format(self.id, self.name, self.price)
Session = sessionmaker(bind=engine)
session = Session()
results1 = session.query(Book).filter(Book.name == None).all()
print('Null Test: ')
for result in results1:
    print(result)
results2 = session.query(Book).filter(Book.name != None).all()
print('Not null Test: ')
for result in results2:
    print(result)

打印：

Null Test: 
Book(id:11, name:None， price：33.0)
Not null Test: 
Book(id:1, name:Name 1， price：45.0)
Book(id:2, name:Name 2， price：49.0)
Book(id:3, name:Name 3， price：14.0)
Book(id:4, name:Name 4， price：16.0)
Book(id:5, name:Name 5， price：17.0)
Book(id:6, name:Name 6， price：3.0)
Book(id:7, name:Name 7， price：25.0)
Book(id:8, name:Name 8， price：16.0)
Book(id:9, name:Name 9， price：32.0)
Book(id:10, name:Name 10， price：22.0)

还可以通过is_()和isnot()方法实现：

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy import create_engine, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_demo'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50))
    price = Column(Float)
    def __str__(self):
        return 'Book(id:{}, name:{}， price：{})'.format(self.id, self.name, self.price)
Session = sessionmaker(bind=engine)
session = Session()
results1 = session.query(Book).filter(Book.name.is_(None)).all()
print('Null Test: ')
for result in results1:
    print(result)
results2 = session.query(Book).filter(Book.name.isnot(None)).all()
print('NNot null Test: ')
for result in results2:
    print(result)

and测试：

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy import create_engine, and_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_demo'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50))
    price = Column(Float)
    def __str__(self):
        return 'Book(id:{}, name:{}， price：{})'.format(self.id, self.name, self.price)
Session = sessionmaker(bind=engine)
session = Session()
results1 = session.query(Book).filter(Book.name == 'Name 1', Book.price <= 50).all()
print('And Test 1: ')
for result in results1:
    print(result)
results2 = session.query(Book).filter(and_(Book.name == 'Name 1', Book.price <= 50)).all()
print('And Test 2: ')
for result in results2:
    print(result)
results3 = session.query(Book).filter(Book.name == 'Name 1').filter(Book.price <= 50).all()
print('And Test 3: ')
for result in results3:
    print(result)

打印：

And Test 1: 
Book(id:1, name:Name 1， price：45.0)
And Test 2: 
Book(id:1, name:Name 1， price：45.0)
And Test 3: 
Book(id:1, name:Name 1， price：45.0)

and可以通过3种方式实现，显然，第一种方法更简单易用。

or测试：

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy import create_engine, or_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_demo'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(50))
    price = Column(Float)
    def __str__(self):
        return 'Book(id:{}, name:{}， price：{})'.format(self.id, self.name, self.price)
Session = sessionmaker(bind=engine)
session = Session()
results = session.query(Book).filter(or_(Book.name == 'Name 1', Book.price >= 30)).all()
for result in results:
    print(result)

打印：

Book(id:1, name:Name 1， price：45.0)
Book(id:2, name:Name 2， price：49.0)
Book(id:9, name:Name 9， price：32.0)
Book(id:11, name:None， price：33.0)

or_()方法中除了放多个过滤条件，还可以只放一个过滤条件。

二、ORM模型的外键约束 1.ORM建立外键关系

在Mysql中，外键可以让表之间的关系更加紧密，SQLAlchemy也支持外键，通过ForeignKey类来实现，并且可以指定表的外键约束。

创建两个存在外键关系的表示例如下：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30))
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id'))
Base.metadata.drop_all()
Base.metadata.create_all()

运行成功即创建表。

如果查看数据库表，未发现外键，可能是因为MySQL默认存储引擎为MyISAM，不支持外键，需要改成InnoDB，在MySQL配置文件种修改示例如下：
flask orm MySQL change engine

要注意，修改之后，需要重启MySQL服务。

再添加数据：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30))
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id'))
# Base.metadata.create_all()
session = sessionmaker(bind=engine)()
user = User(username='Corley')
session.add(user)
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()

查看数据库中，可以看到user和article表中都已经有插入的数据了。
因为添加了外键，此时Article的uid参数必须是User中已经有的id，如果插入Article的数据的uid在User中不存在对应的id，就会报错、不能插入。

2.ORM数据库外键约束

外键约束有4种：

RESTRICT
严格模式，若子表中有父表对应的关联数据，删除父表对应数据，会阻止删除，也是默认项。 NO ACTION
在MySQL中，同RESTRICT。 CASCADE
级联模式，父表操作后，对应子表关联的数据也进行操作。 SET NULL
置空模式，父表被操作之后，子表对应的外键字段被置空。

restrict测试：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30))
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='RESTRICT'))
Base.metadata.drop_all()
Base.metadata.create_all()
session = sessionmaker(bind=engine)()
user = User(username='Corley')
session.add(user)
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()

运行后，删除user表数据：

delete from user where id = 1;

打印：

ERROR 1451 (23000): Cannot delete or update a parent row: a foreign key constraint fails (`flask_orm`.`article`, CONSTRAINT `article_ibfk_1` FOREIGN KEY (`uid`) REFERENCES `user` (`id`))

显然，提示有外键约束，不能删除数据。

通过ORM删除数据库表测试：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30))
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='RESTRICT'))
Base.metadata.drop_all()
Base.metadata.create_all()
Session = sessionmaker(bind=engine)
session = Session()
user = User(username='Corley')
session.add(user)
session.commit()
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()
user = session.query(User).filter(User.username == 'Corley').all()[0]
session.delete(user)
session.commit()

打印：

Traceback (most recent call last):
  File "D:\.virtualenvs\Test-gftU5mTd\lib\site-packages\mysql\connector\connection_cext.py", line 489, in cmd_query
    raw_as_string=raw_as_string)
_mysql_connector.MySQLInterfaceError: Cannot delete or update a parent row: a foreign key constraint fails (`flask_orm`.`article`, CONSTRAINT `article_ibfk_1` FOREIGN KEY (`uid`) REFERENCES `user` (`id`))
During handling of the above exception, another exception occurred:
......
    raw_as_string=self._raw_as_string)
  File "D:\.virtualenvs\Test-gftU5mTd\lib\site-packages\mysql\connector\connection_cext.py", line 492, in cmd_query
    sqlstate=exc.sqlstate)
sqlalchemy.exc.IntegrityError: (mysql.connector.errors.IntegrityError) 1451 (23000): Cannot delete or update a parent row: a foreign key constraint fails (`flask_orm`.`article`, CONSTRAINT `article_ibfk_1` FOREIGN KEY (`uid`) REFERENCES `user` (`id`))
[SQL: DELETE FROM user WHERE user.id = %(id)s]
[parameters: {'id': 1}]
(Background on this error at: http://sqlalche.me/e/gkpj)

显然，此时报错，不允许删除有外键约束的数据。

cascade测试：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30))
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='CASCADE'))
Base.metadata.drop_all()
Base.metadata.create_all()
session = sessionmaker(bind=engine)()
user = User(username='Corley')
session.add(user)
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()

运行后，再删除user表数据：

delete from user where id = 1;

打印：

Query OK, 1 row affected (0.01 sec)

显然，此时删除成功，再查询user表：

select * from user;

打印：

Empty set (0.00 sec)

再查询article表：

select * from article;

打印：

Empty set (0.00 sec)

显然，此时删除user表中数据，article表的数据也被同步删除。

用ORM删除数据测试：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30))
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='CASCADE'))
Base.metadata.drop_all()
Base.metadata.create_all()
Session = sessionmaker(bind=engine)
session = Session()
user = User(username='Corley')
session.add(user)
session.commit()
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()
user = session.query(User).filter(User.username == 'Corley').all()[0]
session.delete(user)
session.commit()

运行后查询数据库与之前结果一致。

set null测试：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30))
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='SET NULL'))
Base.metadata.drop_all()
Base.metadata.create_all()
session = sessionmaker(bind=engine)()
user = User(username='Corley')
session.add(user)
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()

运行后，删除user表数据：

delete from user where id = 1;

打印：

Query OK, 1 row affected (0.01 sec)

显示删除成功，此时查询user表：

select * from user;

打印：

Empty set (0.00 sec)

再查询article表：

select * from article;

打印：

+----+--------+-----------------------------+------+
| id | title  | content                     | uid  |
+----+--------+-----------------------------+------+
|  1 | Python | 人生苦短，我用Python         | NULL |
+----+--------+-----------------------------+------+
1 row in set (0.00 sec)

显然，此时当删除user表中的数据，article表中与其关联的数据被设为null。

用ORM删除数据测试：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30))
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='SET NULL'))
Base.metadata.drop_all()
Base.metadata.create_all()
Session = sessionmaker(bind=engine)
session = Session()
user = User(username='Corley')
session.add(user)
session.commit()
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()
user = session.query(User).filter(User.username == 'Corley').all()[0]
session.delete(user)
session.commit()

运行后查询数据库与之前一致。

3.存在外键时数据的查询

查询数据示例如下：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30))
    def __str__(self):
        return 'User(id: {}, username: {})'.format(self.id, self.username)
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='SET NULL'))
Base.metadata.drop_all()
Base.metadata.create_all()
Session = sessionmaker(bind=engine)
session = Session()
user = User(username='Corley')
session.add(user)
session.commit()
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()
article = session.query(Article).first()
uid = article.uid
user = session.query(User).get(uid)
print(user)

打印：

User(id: 1, username: Corley)

显然，这种方式先通过Article找到用户的uid，再通过uid查找用户，显得很麻烦，可以简化：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30))
    def __str__(self):
        return 'User(id: {}, username: {})'.format(self.id, self.username)
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='SET NULL'))
Base.metadata.drop_all()
Base.metadata.create_all()
Session = sessionmaker(bind=engine)
session = Session()
user = User(username='Corley')
session.add(user)
session.commit()
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()
user = session.query(User).filter(User.id == Article.uid).first()
print(user)

查询结果是一样的。

三、Flask数据库表关系

表之间的关系有三种：

一对一一对多多对多

SQLAlchemy中的ORM可以实现这三种关系。

1.一对多关系

例如：
一个人可以有多辆车，但是一辆车只能对应一个人。

测试如下：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30), nullable=False)
    articles = relationship("Article")
    def __str__(self):
        return 'User(id: {}, username: {})'.format(self.id, self.username)
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='SET NULL'))
    author = relationship('User')
    def __str__(self):
        return 'User(id: {}, title: {}, content: {}, uid: {})'.format(self.id, self.title, self.content, self.uid)
Base.metadata.drop_all()
Base.metadata.create_all()
Session = sessionmaker(bind=engine)
session = Session()
user = User(username='Corley')
session.add(user)
session.commit()
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()
article = session.query(Article).first()
print(article)
print(article.author)
user = session.query(User).first()
print(user)
for article in user.articles:
    print(article)

打印：

User(id: 1, title: Python, content: 人生苦短，我用Python, uid: 1)
User(id: 1, username: Corley)
User(id: 1, username: Corley)
User(id: 1, title: Python, content: 人生苦短，我用Python, uid: 1)

显然，此时根据文章查询作者和根据作者查询文章都能正常查询，因为作者和文章时一对多的关系，所以通过作者查询文章得到的是一个列表，可以遍历列表来获取每一篇文章；
同时，表之间的关系是建立在两个表之间存在外键约束的前提下的，如果未建立外键约束，但在两个表之间建立一对多的关系，会出错。

联合表添加单条数据测试：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30), nullable=False)
    articles = relationship("Article")
    def __str__(self):
        return 'User(id: {}, username: {})'.format(self.id, self.username)
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='SET NULL'))
    author = relationship('User')
    def __str__(self):
        return 'User(id: {}, title: {}, content: {}, uid: {})'.format(self.id, self.title, self.content, self.uid)
Base.metadata.drop_all()
Base.metadata.create_all()
Session = sessionmaker(bind=engine)
session = Session()
user = User(username='Corley')
session.add(user)
session.commit()
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()
user = User(username='Lord')
article = Article(title='Java', content='Spring')
article.author = user
session.add(article)
session.commit()

运行后，查询user表：

select * from user;

打印：

+----+----------+       
| id | username |       
+----+----------+       
|  1 | Corley   |       
|  2 | Lord     |       
+----+----------+       
2 rows in set (0.01 sec)

再查询article表：

select * from article;

打印：

+----+--------+----------------------+------+
| id | title  | content              | uid  |
+----+--------+----------------------+------+
|  1 | Python | 人生苦短，我用Python  |    1 |
|  2 | Java   | Spring               |    2 |
+----+--------+----------------------+------+
2 rows in set (0.00 sec)

显然，两条数据都插入成功，并形成映射关系。

添加多条数据测试：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30), nullable=False)
    articles = relationship("Article")
    def __str__(self):
        return 'User(id: {}, username: {})'.format(self.id, self.username)
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='SET NULL'))
    author = relationship('User')
    def __str__(self):
        return 'User(id: {}, title: {}, content: {}, uid: {})'.format(self.id, self.title, self.content, self.uid)
Base.metadata.drop_all()
Base.metadata.create_all()
Session = sessionmaker(bind=engine)
session = Session()
user = User(username='Corley')
session.add(user)
session.commit()
article = Article(title='Python', content='人生苦短，我用Python', uid=1)
session.add(article)
session.commit()
user = User(username='Jessica')
article1 = Article(title='Java', content='Spring')
article2 = Article(title='C', content='Clang')
article1.author = user
article2.author = user
session.add_all([article1, article2])
session.commit()

运行后，查询user表：

select * from user;

打印：

+----+----------+
| id | username |
+----+----------+
|  1 | Corley   |
|  2 | Jessica  |
+----+----------+
2 rows in set (0.00 sec)

再查询article表：

select * from article;

打印：

+----+--------+----------------------+------+
| id | title  | content              | uid  |
+----+--------+----------------------+------+
|  1 | Python | 人生苦短，我用Python  |    1 |
|  2 | Java   | Spring               |    2 |
|  3 | C      | Clang                |    2 |
+----+--------+----------------------+------+
3 rows in set (0.00 sec)

显然，此时也通过一对多的映射将一个user和对应的两个article插入数据库。

之前都是通过在User和Article模型中都定义关系映射来实现一对多的关系，但是也可以只在一个模型中定义relationship来实现一对多的关系，此时需要定义反向访问属性backref，示例如下：

from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
# 数据库连接配置
HOSTNAME = '127.0.0.1'
PORT = 3306
USERNAME = 'root'
PASSWORD = 'root'
DATABASE = 'flask_orm'
DB_URL = 'mysql+mysqlconnector://{}:{}@{}:{}/{}?charset=utf8'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
engine = create_engine(DB_URL)
Base = declarative_base(engine)
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(30), nullable=False)
    articles = relationship("Article", backref='author')        # 定义反向访问属性
    def __str__(self):
        return 'User(id: {}, username: {})'.format(self.id, self.username)
class Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50))
    content = Column(Text, nullable=False)
    uid = Column(Integer, ForeignKey('user.id', ondelete='SET NULL'))
    # author = relationship('User')
    def __str__(self):
        return 'User(id: {}, title: {}, content: {}, uid: {})'.format(self.id, self.title, self.content, self.uid)
# Base.metadata.drop_all()
Base.metadata.create_all()
Session = sessionmaker(bind=engine)
session = Session()
article = session.query(Article).first()
print(article)
print(article.author)
user = session.query(User).first()
print(user)
for article in user.articles:
    print(article)

打印：

User(id: 1, title: Python, content: 人生苦短，我用Python, uid: 1)
User(id: 1, username: Corley)
User(id: 1, username: Corley)
User(id: 1, title: Python, content: 人生苦短，我用Python, uid: 1)
User(id: 4, title: Python, content: 人生苦短，我用Python, uid: 1)

显然，此时可以互相查询；
在Article模型中未定义关系，只在User模型中定义了关系，同时在定义关系时定义了backref反向访问属性，来保证不仅能通过User来查询Article，还要能通过Article来查询User。

作者：cutercorley

flask python全栈关系外键 orm Python

1024 个赞