ORM表关系、排序、分页和高级查询

• 一、数据库排序
• 二、limit、offset和切片
• 三、高级查询和子查询
• 1.group_by
• 2.having
• 3.子查询

一、数据库排序

在ORM中排序的方式有2种：

• order_by()
  在查询结果时，使用order_by()方法，可以指定根据表中的某个字段进行排序。
• 模型定义时声明
  有时候，不想每次在查询的时候都指定排序的方式，可以在定义模型的时候就指定默认排序的方式。

默认情况下是升序，还可以指定按降序方式输出结果。

生成数据：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from random import randint

# 数据库连接配置
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 Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50), nullable=False)
    length = Column(Integer)

    def __str__(self):
        return 'User(id: {}, name: {}, length: {})'.format(self.id, self.title, self.length)


Base.metadata.drop_all()
Base.metadata.create_all()

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

for i in range(10):
    article = Article(title='Title %d' % i, length=randint(100, 10000))
    session.add(article)

session.commit()

运行后即插入数据。

此时用order_by()方法对查询结果排序，测试如下：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from random import randint

# 数据库连接配置
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 Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50), nullable=False)
    length = Column(Integer)

    def __str__(self):
        return 'User(id: {}, name: {}, length: {})'.format(self.id, self.title, self.length)


# Base.metadata.drop_all()
Base.metadata.create_all()

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

articles = session.query(Article).order_by(Article.length).all()
for article in articles:
    print(article)

打印：

User(id: 4, name: Title 3, length: 143)
User(id: 6, name: Title 5, length: 1079)
User(id: 3, name: Title 2, length: 1323)
User(id: 7, name: Title 6, length: 1589)
User(id: 1, name: Title 0, length: 1604)
User(id: 2, name: Title 1, length: 2187)
User(id: 8, name: Title 7, length: 2933)
User(id: 10, name: Title 9, length: 3556)
User(id: 5, name: Title 4, length: 3829)
User(id: 9, name: Title 8, length: 8520)

显然，是按length属性从小到大排列的，也可以得到order_by()方法默认是升序排序。

如果想要降序排列，可以采用如下两种方式：

• 使用desc()方法

order_by(Article.length.desc()).all()

• 字段前面加符号-

order_by(-Article.length).all()

其一测试如下：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from random import randint

# 数据库连接配置
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 Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50), nullable=False)
    length = Column(Integer)

    def __str__(self):
        return 'User(id: {}, name: {}, length: {})'.format(self.id, self.title, self.length)


# Base.metadata.drop_all()
Base.metadata.create_all()

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

articles = session.query(Article).order_by(Article.length.desc()).all()
for article in articles:
    print(article)

打印：

User(id: 9, name: Title 8, length: 8520)
User(id: 5, name: Title 4, length: 3829)
User(id: 10, name: Title 9, length: 3556)
User(id: 8, name: Title 7, length: 2933)
User(id: 2, name: Title 1, length: 2187)
User(id: 1, name: Title 0, length: 1604)
User(id: 7, name: Title 6, length: 1589)
User(id: 3, name: Title 2, length: 1323)
User(id: 6, name: Title 5, length: 1079)
User(id: 4, name: Title 3, length: 143)

这是通过在查询结果时使用order_by()方法来实现的，如果要想在查询时就默认以某一种方式排序，可以在建立模型时定义，如下：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from random import randint

# 数据库连接配置
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 Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50), nullable=False)
    length = Column(Integer)

    def __str__(self):
        return 'User(id: {}, name: {}, length: {})'.format(self.id, self.title, self.length)

    __mapper_args__ = {
        'order_by':length
    }


# Base.metadata.drop_all()
Base.metadata.create_all()

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

articles = session.query(Article).all()
for article in articles:
    print(article)

输出与之前默认升序一样的结果，显然此时不再需要在查询时使用order_by()方法，还可以降序，如下：

__mapper_args__ = {
	'order_by':length.desc()
}

或

__mapper_args__ = {
	'order_by':-length
}

显然，在定义模型时就定义默认排序更好，因为如果在查询时再定义排序，会降低查询效率。

二、limit、offset和切片

• limit
  指定返回行的最大记录数目。
• offset
  指定第一个返回记录行的偏移量。
• 切片
  对Query对象使用切片操作，来获取想要的数据。

limit测试：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from random import randint

# 数据库连接配置
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 Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50), nullable=False)
    length = Column(Integer)

    def __str__(self):
        return 'User(id: {}, name: {}, length: {})'.format(self.id, self.title, self.length)

    __mapper_args__ = {
        'order_by':length
    }


# Base.metadata.drop_all()
Base.metadata.create_all()

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


articles = session.query(Article).limit(3).all()
for article in articles:
    print(article)

打印：

User(id: 9, name: Title 8, length: 1895)
User(id: 1, name: Title 0, length: 2038)
User(id: 7, name: Title 6, length: 2456)

显然，此时只查询了前3条数据。

offset测试：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from random import randint

# 数据库连接配置
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 Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50), nullable=False)
    length = Column(Integer)

    def __str__(self):
        return 'User(id: {}, name: {}, length: {})'.format(self.id, self.title, self.length)

    __mapper_args__ = {
        'order_by':length
    }


# Base.metadata.drop_all()
Base.metadata.create_all()

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


articles = session.query(Article).offset(3).limit(3).all()
for article in articles:
    print(article)

打印：

User(id: 3, name: Title 2, length: 4884)
User(id: 6, name: Title 5, length: 5197)
User(id: 8, name: Title 7, length: 8874)

上面查询了从第4条开始的连续3条数据；
显然，offset的参数是从0开始计数的。

还可以多个查询条件同时使用：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from random import randint

# 数据库连接配置
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 Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50), nullable=False)
    length = Column(Integer)

    def __str__(self):
        return 'User(id: {}, name: {}, length: {})'.format(self.id, self.title, self.length)


# Base.metadata.drop_all()
Base.metadata.create_all()

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


articles = session.query(Article).order_by(Article.length.desc()).offset(3).limit(3).all()
for article in articles:
    print(article)

打印：

User(id: 5, name: Title 4, length: 8897)
User(id: 8, name: Title 7, length: 8874)
User(id: 6, name: Title 5, length: 5197)

只能先用order_by()排序，再用limit()和offset()取数据，而不能颠倒两者的顺序。

切片测试如下：

from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from random import randint

# 数据库连接配置
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 Article(Base):
    __tablename__ = 'article'
    id = Column(Integer, primary_key=True, autoincrement=True)
    title = Column(String(50), nullable=False)
    length = Column(Integer)

    def __str__(self):
        return 'User(id: {}, name: {}, length: {})'.format(self.id, self.title, self.length)


# Base.metadata.drop_all()
Base.metadata.create_all()

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


articles = session.query(Article).all()[2:5]
for article in articles:
    print(article)

打印：

User(id: 3, name: Title 2, length: 4884)
User(id: 4, name: Title 3, length: 9693)
User(id: 5, name: Title 4, length: 8897)

用limit()和切片都可以截取部分数据，其区别在于：
用limit()是在查询时就限定了查询的结果，即查询的结果并不包括所有结果，而是指定的部分；
而切片是现将所有符合的结果查询出来，得到的是一个列表，再对列表切片；
显然，用limit()方法的效率更高。

三、高级查询和子查询

1.group_by

group_by()指定根据某个字段分组。

创建数据库表并添加数据：

from sqlalchemy import Column, Integer, String, Enum
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from random import *

# 数据库连接配置
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)


def generate_random_str(randomlength=6):
    '''生成指定长度的随机字符串'''
    random_str = ''
    base_str = 'ABCDEFGHIGKLMNOPQRSTUVWXYZabcdefghigklmnopqrstuvwxyz0123456789'
    length = len(base_str) - 1
    for i in range(randomlength):
        random_str += base_str[randint(0, length)]
    return random_str


class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), nullable=False)
    gender = Column(Enum('男', '女'))
    age = Column(Integer)

    def __str__(self):
        return 'User(id: {}, username: {}, gender: {}, age: {})'.format(self.id, self.username, self.gender, self.age)


Base.metadata.drop_all()
Base.metadata.create_all()

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


for _ in range(20):
    user = User(username=generate_random_str(), gender=choice(['男', '女']), age=randint(5, 40))
    session.add(user)

session.commit()

用group_by()进行查询：

from sqlalchemy import Column, Integer, String, Enum
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from random import *

# 数据库连接配置
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(50), nullable=False)
    gender = Column(Enum('男', '女'))
    age = Column(Integer)

    def __str__(self):
        return 'User(id: {}, username: {}, gender: {}, age: {})'.format(self.id, self.username, self.gender, self.age)


# Base.metadata.drop_all()
Base.metadata.create_all()

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

results = session.query(User.gender).group_by(User.gender).all()
for result in results:
    print(result)

打印：

('男',)
('女',)

很显然，query()和group_by()的参数所包含的字段要保持一致。

可以对每个类别进行计数，如下：

from sqlalchemy import Column, Integer, String, Enum
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_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(50), nullable=False)
    gender = Column(Enum('男', '女'))
    age = Column(Integer)

    def __str__(self):
        return 'User(id: {}, username: {}, gender: {}, age: {})'.format(self.id, self.username, self.gender, self.age)


# Base.metadata.drop_all()
Base.metadata.create_all()

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

results = session.query(User.gender, func.count(User.id)).group_by(User.gender).all()
for result in results:
    print(result)

打印：

('男', 12)
('女', 8)

显然，查到了性别为男、女的人数。

2.having

having()对通过group_by()分类后的结果进一步过滤。
显然，having()中的字段也应该与group_by()中的字段保持一致。

根据年龄分组后再根据年龄条件过滤：

from sqlalchemy import Column, Integer, String, Enum
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_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(50), nullable=False)
    gender = Column(Enum('男', '女'))
    age = Column(Integer)

    def __str__(self):
        return 'User(id: {}, username: {}, gender: {}, age: {})'.format(self.id, self.username, self.gender, self.age)


# Base.metadata.drop_all()
Base.metadata.create_all()

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

results = session.query(User.age, func.count(User.id)).group_by(User.age).having(User.age > 18).all()
for result in results:
    print(result)

打印：

(20, 1)
(21, 2)
(22, 1)
(32, 1)
(34, 1)
(36, 1)
(39, 3)

显然，得到了大于18的各个年龄阶段的人数。

3.子查询

MySQL中有子查询，SQLAlchemy也支持子查询，一般过程是先构造子查询，再将子查询放到父查询中。

构造数据：

from random import *
from sqlalchemy import Column, Integer, String, Enum
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)


def generate_random_str(randomlength=6):
    '''生成指定长度的随机字符串'''
    random_str = ''
    base_str = 'ABCDEFGHIGKLMNOPQRSTUVWXYZabcdefghigklmnopqrstuvwxyz0123456789'
    length = len(base_str) - 1
    for i in range(randomlength):
        random_str += base_str[randint(0, length)]
    return random_str


class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), nullable=False)
    gender = Column(Enum('男', '女'))
    age = Column(Integer)
    city = Column(Enum('Beijing', 'New York', 'London', 'Paris', 'Tokyo'))

    def __str__(self):
        return 'User(id: {}, username: {}, gender: {}, age: {}, city: {})'.format(self.id, self.username, self.gender, self.age, self.city)


Base.metadata.drop_all()
Base.metadata.create_all()

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

for _ in range(50):
    user = User(username=generate_random_str(), gender=choice(['男', '女']), age=randint(5, 40), city=choice(['Beijing', 'New York', 'London', 'Paris', 'Tokyo']))
    session.add(user)

session.commit()

运行即创建数据成功。

现在有一个需求，查询和id为3的user在同一个城市的user，可以先用一条语句查询出id为3的用户所在的城市，然后再根据城市进行第二次查询：

from random import *
from sqlalchemy import Column, Integer, String, Enum
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(50), nullable=False)
    gender = Column(Enum('男', '女'))
    age = Column(Integer)
    city = Column(Enum('Beijing', 'New York', 'London', 'Paris', 'Tokyo'))

    def __str__(self):
        return 'User(id: {}, username: {}, gender: {}, age: {}, city: {})'.format(self.id, self.username, self.gender, self.age, self.city)


# Base.metadata.drop_all()
Base.metadata.create_all()

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

user = session.query(User).filter(User.id == 3).first()
results = session.query(User).filter(User.city == user.city).all()
for result in results:
    print(result)

打印：

User(id: 3, username: gTq26G, gender: 女, age: 8, city: Paris)
User(id: 10, username: h7hs1d, gender: 男, age: 35, city: Paris)
User(id: 14, username: xmbm6H, gender: 男, age: 31, city: Paris)
User(id: 20, username: Gq0AlG, gender: 男, age: 39, city: Paris)
User(id: 25, username: ksGOnY, gender: 女, age: 32, city: Paris)
User(id: 27, username: H8Amxe, gender: 女, age: 34, city: Paris)
User(id: 28, username: 9Vmt4P, gender: 女, age: 9, city: Paris)
User(id: 31, username: rM3Qoe, gender: 男, age: 20, city: Paris)
User(id: 32, username: 1XEtX3, gender: 男, age: 35, city: Paris)
User(id: 33, username: 3ZzS4b, gender: 男, age: 13, city: Paris)
User(id: 34, username: naCdU4, gender: 女, age: 35, city: Paris)
User(id: 35, username: iG7Yfp, gender: 女, age: 27, city: Paris)
User(id: 36, username: 12kwgW, gender: 男, age: 34, city: Paris)
User(id: 37, username: Dwl6Ba, gender: 女, age: 18, city: Paris)
User(id: 47, username: ehe80s, gender: 女, age: 7, city: Paris)
User(id: 50, username: MdxxLe, gender: 女, age: 33, city: Paris)

显然，打印出了所有符合条件的结果，但是这是多个语句查询出的，显得很麻烦，可以用子查询，如下：

from sqlalchemy import Column, Integer, String, Enum
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(50), nullable=False)
    gender = Column(Enum('男', '女'))
    age = Column(Integer)
    city = Column(Enum('Beijing', 'New York', 'London', 'Paris', 'Tokyo'))

    def __str__(self):
        return 'User(id: {}, username: {}, gender: {}, age: {}, city: {})'.format(self.id, self.username, self.gender, self.age, self.city)


# Base.metadata.drop_all()
Base.metadata.create_all()

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

sub = session.query(User.city.label('city')).filter(User.id == 3).subquery()
results = session.query(User).filter(User.city == sub.c.city).all()
for result in results:
    print(result)

可以看到，一个查询如果想要变为子查询，需要通过subquery()方法实现，变成子查询后，通过子查询.c属性来访问查询出来的列。