《SQLAlchemy 2实战》习题解答SQLAlchemy 2 In Practice - Solutions to the Exercises

在《SQLAlchemy 2实战》系列文章的结尾，本文包含了所有习题的解答。如果您希望支持我的工作，欢迎通过我的商店或亚马逊直接购买本书，非常感谢！

以下是本书内容的概要：

第2章

1. 按字母顺序排列，列出1983年生产的最早三款产品。

>>> q = (select(Product)
        .where(Product.year == 1983)
        .order_by(Product.name)
        .limit(3))
>>> session.scalars(q).all()
[Product(17, "Apple IIe"), Product(85, "Aquarius"), Product(26, "Atari 1200XL")]

2. 使用“Z80”CPU或其克隆产品的所有产品。假设基于该CPU的产品都会在cpu列中包含“Z80”字样。

>>> q = (select(Product)
        .where(Product.cpu.like('%Z80%')))
>>> session.scalars(q).all()
[ ... 63 results ... ]

3. 使用“Z80”或“6502”CPU或其克隆产品，且生产于1990年之前的产品，按名称字母顺序排序。

>>> q = (select(Product)
        .where(
            or_(Product.cpu.like('%Z80%'), Product.cpu.like('%6502%')),
            Product.year < 1990)
        .order_by(Product.name))
>>> session.scalars(q).all()
[ ... 90 results ... ]

4. 在1980年代生产过产品的制造商。

>>> q = (select(Product.manufacturer)
        .where(Product.year.between(1980, 1989))
        .distinct())
>>> session.scalars(q).all()
[ ... 65 results ... ]

5. 名称以字母“T”开头的制造商，按字母顺序排序。

>>> q = (select(Product.manufacturer)
        .where(Product.manufacturer.like('T%'))
        .order_by(Product.manufacturer)
        .distinct())
>>> session.scalars(q).all()
['Tangerine Computer Systems', 'Technosys', 'Tesla', 'Texas Instruments',
'Thomson', 'Timex Sinclair', 'Tomy', 'Tsinghua University']

6. 克罗地亚生产产品的最早和最晚年份，以及对应的生产数量。

>>> q = select(
        func.min(Product.year), func.max(Product.year),
        func.count(Product.id)
    ).where(Product.country == 'Croatia')
>>> session.execute(q).first()
(1981, 1984, 4)

7. 每年生产的产品数量，结果从产品最多的年份到最少的年份排列，未生产过的年份无需显示。

>>> product_count = func.count(Product.id).label(None)
>>> q = (select(Product.year, product_count)
        .group_by(Product.year)
        .order_by(product_count.desc()))
>>> session.execute(q).all()
[(1983, 24), (1984, 21), (1985, 21), (1982, 17), (1986, 11), (1980, 10),
(1979, 9), (1977, 7), (1987, 6), (1981, 6), (1990, 5), (1989, 4), (1988, 2),
(1978, 2), (1969, 1), (1995, 1), (1992, 1), (1991, 1)]

8. 美国（注意这些产品的国家字段设置为USA）的制造商数量。

>>> q = (select(func.count(Product.manufacturer.distinct()))
        .where(Product.country == 'USA'))
>>> session.scalar(q)
17

第3章

1. IBM和Texas Instruments制造的产品列表。

>>> q = (select(Product)
        .join(Product.manufacturer)
        .where(or_(
            Manufacturer.name == 'IBM',
            Manufacturer.name == 'Texas Instruments')))
>>> session.scalars(q).all()
[Product(75, "PCjr"), Product(76, "IBM PS/1"), Product(132, "TI-99/4"),
Product(133, "TI-99/4A")]

另一种方案是使用in_()操作符查询列：

>>> q = (select(Product)
        .join(Product.manufacturer)
        .where(Manufacturer.name.in_(['IBM', 'Texas Instruments'])))

2. 在巴西运营的制造商。

>>> q = (select(Manufacturer)
        .join(Manufacturer.products)
        .where(Product.country == 'Brazil')
        .distinct())
>>> session.scalars(q).all()
[Manufacturer(32, "Gradiente"), Manufacturer(46, "Comércio de Componentes Eletrônicos"),
Manufacturer(47, "Microdigital Eletronica"), Manufacturer(59, "Prológica")]

另一种方案是用group_by()替代distinct()：

>>> q = (select(Manufacturer)
        .join(Manufacturer.products)
        .where(Product.country == 'Brazil')
        .group_by(Manufacturer))

3. 制造商名称中包含“Research”字样的产品。

>>> q = (select(Product)
        .join(Product.manufacturer)
        .where(Manufacturer.name.like('%Research%')))
>>> session.scalars(q).all()
[Product(125, "ZX80"), Product(126, "ZX81"), Product(127, "ZX Spectrum"),
Product(128, "Sinclair QL")]

4. 生产基于Z80 CPU或其克隆产品的制造商。

>>> q = (select(Manufacturer)
        .join(Manufacturer.products)
        .where(Product.cpu.like('%Z80%'))
        .distinct())
>>> session.scalars(q).all()
[ ... 39 results ... ]

5. 未生产基于6502 CPU或其克隆产品的制造商。

>>> q = (select(Manufacturer)
        .join(Manufacturer.products)
        .where(not_(Product.cpu.like('%8502%')))
        .distinct())
>>> session.scalars(q).all()
[ ... 76 results ... ]

6. 制造商及其首款产品上市年份，按年份排序。

>>> first_year = func.min(Product.year).label(None)
>>> q = (select(Manufacturer, first_year)
        .join(Manufacturer.products)
        .group_by(Manufacturer)
        .order_by(first_year))
>>> session.execute(q).all()
[ ... 76 results ... ]

7. 目录中拥有3至5个产品的制造商。

>>> q = (select(Manufacturer)
        .join(Manufacturer.products)
        .group_by(Manufacturer)
        .having(func.count(Product.id).between(3, 5)))
>>> session.scalars(q).all()
[Manufacturer(54, "Tangerine Computer Systems"), Manufacturer(63, "Sinclair Research"),
Manufacturer(60, "VEB Robotron"), Manufacturer(62, "Sharp"),
Manufacturer(44, "Memotech"), Manufacturer(9, "Atari Corporation"),
Manufacturer(57, "Pravetz"), Manufacturer(20, "Didaktik"), Manufacturer(56, "Philips")]

8. 运营超过5年的制造商。

>>> q = (select(Manufacturer)
        .join(Manufacturer.products)
        .group_by(Manufacturer)
        .having(func.max(Product.year) - func.min(Product.year) > 5))
>>> session.scalars(q).all()
[Manufacturer(34, "IBM"), Manufacturer(52, "Radio Shack"),
Manufacturer(14, "Commodore"), Manufacturer(2, "Amstrad"), Manufacturer(62, "Sharp"),
Manufacturer(9, "Atari Corporation"), Manufacturer(30, "Fujitsu"),
Manufacturer(1, "Acorn Computers Ltd"), Manufacturer(5, "Apple Computer"),
Manufacturer(8, "Atari, Inc.")]

第4章

1. 在英国或美国生产的产品。

>>> q = (select(Product)
        .join(Product.countries)
        .where(Country.name.in_(['UK', 'USA']))
        .distinct())

注意：此查询需要使用distinct()子句来消除两国联合生产时的重复项。

2. 不在英国或美国生产的产品。包括与其他国家联合生产于英国和/或美国的产品。

>>> q = (select(Product)
        .join(Product.countries)
        .where(not_(Country.name.in_(['UK', 'USA'])))
        .distinct())
>>> session.scalars(q).all()
[ ... 70 results ... ]

不使用in_()操作符的替代方案：

>>> q = (select(Product)
        .join(Product.countries)
        .where(Country.name != 'UK', Country.name != 'USA')
        .distinct())

3. 生产基于Z80 CPU或其克隆产品的国家。

>>> q = (select(Country)
        .join(Country.products)
        .where(Product.cpu.like('%Z80%'))
        .distinct())
>>> session.scalars(q).all()
[Country(11, "Japan"), Country(12, "Brazil"), Country(7, "Belgium"),
Country(24, "Hungary"), Country(16, "Australia"), Country(4, "Netherlands"),
Country(1, "UK"), Country(3, "USA"), Country(25, "Norway"),
Country(21, "East Germany"), Country(5, "Romania"), Country(22, "Portugal"),
Country(6, "Hong Kong"), Country(9, "USSR"), Country(14, "Sweden"),
Country(8, "Czechoslovakia"), Country(23, "Poland")]

4. 按字母顺序排列，1970年代生产产品的国家。

>>> q = (select(Country)
        .join(Country.products)
        .where(Product.year.between(1970, 1979))
        .order_by(Country.name)
        .distinct())
>>> session.scalars(q).all()
[Country(11, "Japan"), Country(14, "Sweden"), Country(3, "USA")]

5. 产品数量最多的前五个国家。若出现并列情况，按国家字母顺序选择。

>>> product_count = func.count(Product.id).label(None)
>>> q = (select(Country, product_count)
        .join(Country.products)
        .group_by(Country)
        .order_by(product_count.desc(), Country.name)
        .limit(5))
>>> session.execute(q).all()
[(Country(3, "USA"), 51), (Country(1, "UK"), 36), (Country(11, "Japan"), 12),
(Country(6, "Hong Kong"), 6), (Country(22, "Portugal"), 6)]

6. 在英国或美国拥有超过3个产品的制造商。

>>> product_count = func.count(Product.id.distinct()).label(None)
>>> q = (select(Manufacturer, product_count)
        .join(Manufacturer.products)
        .join(Product.countries)
        .where(Country.name.in_(['UK', 'USA']))
        .group_by(Manufacturer)
        .having(product_count > 3))
>>> session.execute(q).all()
[(Manufacturer(1, "Acorn Computers Ltd"), 6), (Manufacturer(2, "Amstrad"), 7),
(Manufacturer(5, "Apple Computer"), 6), (Manufacturer(8, "Atari, Inc."), 7),
(Manufacturer(14, "Commodore"), 10), (Manufacturer(52, "Radio Shack"), 6),
(Manufacturer(63, "Sinclair Research"), 4), (Manufacturer(70, "Timex Sinclair"), 4)]

7. 产品在多个国家有分布的制造商。

>>> country_count = func.count(Country.id.distinct()).label(None)
>>> q = (select(Manufacturer, country_count)
        .join(Manufacturer.products)
        .join(Product.countries)
        .group_by(Manufacturer)
        .having(country_count > 1))
>>> session.execute(q).all()
[(Manufacturer(70, "Timex Sinclair"), 4)]

8. 在英国和美国联合生产的产品。

>>> q = (select(Product)
        .join(Product.countries)
        .where(Country.name.in_(['UK', 'USA']))
        .group_by(Product)
        .having(func.count(Country.id) > 1))
>>> session.execute(q).all()
[(Product(138, "Timex Sinclair 1000"),), (Product(139, "Timex Sinclair 1500"),),
(Product(140, "Timex Sinclair 2048"),), (Product(142, "Timex Computer 2068"),)]

注意：使此查询生效的技巧是where()子句过滤掉非目标国家的任何产品，因此按产品分组后，行数为2的产品必须同时关联这两个国家。

第5章

1. 金额超过300美元的订单，按销售金额从高到低降序排列。

order_total = func.sum(OrderItem.unit_price * OrderItem.quantity).label(None)
q = (select(Order, order_total)
        .join(Order.order_items)
        .group_by(Order)
        .having(order_total > 300)
        .order_by(order_total.desc()))
>>> session.execute(q).all()
[ ... 50 results ... ]

2. 包含一台或多台ZX81计算机的订单。

q = (select(Order)
        .join(Order.order_items)
        .join(OrderItem.product)
        .where(Product.name == 'ZX81'))
>>> session.scalars(q).all()
[ ... 3 results ... ]

可能更高效的双查询方案，但减少了一次连接：

>>> zx81 = session.scalar(
        select(Product)
            .where(Product.name == 'ZX81'))
>>> q = (select(Order)
        .join(Order.order_items)
        .where(OrderItem.product == zx81))

3. 包含Amstrad制造产品的订单。

>>> q = (select(Order)
        .join(Order.order_items)
        .join(OrderItem.product)
        .join(Product.manufacturer)
        .where(Manufacturer.name == 'Amstrad')
        .distinct())
>>> session.scalars(q).all()
[ ... 30 results ... ]

使用一次连接较少的替代版本：

>>> amstrad = session.scalar(
        select(Manufacturer)
            .where(Manufacturer.name == 'Amstrad'))
>>> q = (select(Order)
        .join(Order.order_items)
        .join(OrderItem.product)
        .where(Product.manufacturer == amstrad)
        .distinct())

注意：这里必须使用 distinct() 来删除那些包含该制造商两个及以上产品明细的订单中的重复项。

4. 2022年12月25日下单且包含两个及以上明细的订单。

>>> q = (select(Order)
        .join(Order.order_items)
        .where(Order.timestamp.between(
            datetime(2022, 12, 25), datetime(2022, 12, 26)))
        .group_by(Order)
        .having(func.count(Order.id) >= 2))
>>> session.scalars(q).all()
[ ... 4 results ... ]

请注意，技术上上述查询也会精确到00:00:00.0时捕获12月26日的订单，因为 between() 操作符包含起始和结束值。更准确（但较长）的查询可使用 extract() 函数构建：

>>> q = (select(Order)
        .join(Order.order_items)
        .where(
            func.extract('day', Order.timestamp) == 25,
            func.extract('month', Order.timestamp) == 12,
            func.extract('year', Order.timestamp) == 2022)
        .group_by(Order)
        .having(func.count(Order.id) >= 2))
>>> session.scalars(q).all()

5. 客户的首单和末单日期及时间提示：min() 和 max() 函数可帮助完成此查询。

>>> q = (select(Customer, func.min(Order.timestamp), func.max(Order.timestamp))
        .join(Customer.orders)
        .group_by(Customer))
>>> session.execute(q).all()
[ ... 2754 results ... ]

6. 按销售额降序排列的前5大制造商及其销售总额。

>>> order_total = func.sum(OrderItem.unit_price * OrderItem.quantity).label(None)
>>> q = (select(Manufacturer, order_total)
        .join(Manufacturer.products)
        .join(Product.order_items)
        .group_by(Manufacturer)
        .order_by(order_total.desc())
        .limit(5))
>>> session.execute(q).all()
[(Manufacturer(14, "Commodore"), 281666.6599999996),
(Manufacturer(63, "Sinclair Research"), 122582.61999999928),
(Manufacturer(5, "Apple Computer"), 34169.33000000025),
(Manufacturer(1, "Acorn Computers Ltd"), 14018.28000000003),
(Manufacturer(8, "Atari, Inc."), 3154.7399999999984)]

7. 产品、其平均星级评分及评论数量，按评论数量降序排列。

>>> product_rating = func.avg(ProductReview.rating).label(None)
>>> review_count = func.count(ProductReview.rating).label(None)
>>> q = (select(Product, product_rating, review_count)
        .join(Product.reviews)
        .group_by(Product)
        .order_by(review_count.desc()))
>>> session.execute(q).all()
[ ... 125 results ... ]

上述解决方案未包含无评论的产品。若要包含缺失的产品，需将连接升级为左外连接：

>>> q = (select(Product, product_rating, review_count)
        .join(Product.reviews, isouter=True)
        .group_by(Product)
        .order_by(review_count.desc()))
>>> session.execute(q).all()
[ ... 149 results ... ]

8. 产品及其平均星级评分，仅统计包含书面评论的评分。

>>> product_rating = func.avg(ProductReview.rating).label(None)
>>> q = (select(Product, product_rating)
        .join(Product.reviews)
        .where(ProductReview.comment != None)
        .group_by(Product))
>>> session.execute(q).all()
[ ... 70 results ... ]

9. 2022年Commodore 64计算机各月的平均星级评分。

>>> month = func.extract('month', ProductReview.timestamp).label(None)
>>> year = func.extract('year', ProductReview.timestamp).label(None)
>>> product_rating = func.avg(ProductReview.rating).label(None)
>>> q = (select(year, month, product_rating)
        .join(ProductReview.product)
        .where(Product.name == 'Commodore 64')
        .group_by(year, month)
        .order_by(year, month))
>>> session.execute(q).all()
[(2022, 1, 4.294117647058823), (2022, 2, 3.6551724137931036),
(2022, 3, 3.4375), (2022, 4, 3.975609756097561), (2022, 5, 3.317073170731707),
(2022, 6, 3.6774193548387095), (2022, 7, 3.606060606060606),
(2022, 8, 3.973684210526316), (2022, 9, 3.6666666666666665), (2022, 10, 3.9375),
(2022, 11, 3.8958333333333335), (2022, 12, 3.6)]

如前几题所述，若提前获取产品信息，则可移除连接：

>>> c64 = session.scalar(select(Product).where(Product.name == 'Commodore 64'))
>>> q = (select(year, month, product_rating)
        .where(ProductReview.product == c64)
        .group_by(year, month)
        .order_by(year, month))

10. 客户给予产品的最低和最高星级评分，按客户名称字母顺序排序。

>>> min_rating = func.min(ProductReview.rating).label(None)
>>> max_rating = func.max(ProductReview.rating).label(None)
>>> q = (select(Customer, min_rating, max_rating)
        .join(Customer.product_reviews)
        .group_by(Customer)
        .order_by(Customer.name))
>>> session.execute(q).all()
[ ... 931 results ... ]

11. 制造商及其平均星级评分，从高到低排序。

>>> product_rating = func.avg(ProductReview.rating).label(None)
>>> q = (select(Manufacturer, product_rating)
        .join(Manufacturer.products)
        .join(Product.reviews)
        .group_by(Manufacturer)
        .order_by(product_rating.desc()))
>>> session.execute(q).all()
[ ... 68 results ... ]

上述方案仅报告至少有一个产品被评分的制造商。要包含无任何评分产品的制造商，需将连接升级为左外连接：

>>> q = (select(Manufacturer, product_rating)
        .join(Manufacturer.products, isouter=True)
        .join(Product.reviews, isouter=True)
        .group_by(Manufacturer)
        .order_by(product_rating.desc()))
>>> session.execute(q).all()
[ ... 76 results ... ]

12. 产品国家及其平均星级评分，从高到低排序。

>>> product_rating = func.avg(ProductReview.rating).label(None)
>>> q = (select(Country, product_rating)
        .join(Country.products)
        .join(Product.reviews)
        .group_by(Country)
        .order_by(product_rating.desc()))
>>> session.execute(q).all()
[ ... 23 results ... ]

同上，仅包含至少有一个产品被评分的国家。要加入无评分的国家，需使用左外连接：

>>> q = (select(Country, product_rating)
        .join(Country.products, isouter=True)
        .join(Product.reviews, isouter=True)
        .group_by(Country)
        .order_by(product_rating.desc()))
>>> session.execute(q).all()
[ ... 25 results ... ]

第六章

1. 2020年3月获得超过40次浏览量的博客文章。

>>> q = (select(BlogArticle)
        .join(BlogArticle.views)
        .where(BlogView.timestamp.between(
            datetime(2020, 3, 1), datetime(2020, 4, 1)))
        .group_by(BlogArticle)
        .having(func.count(BlogView.id) > 50))
>>> session.execute(q).all()
[(BlogArticle(143, "Evening however issue"),)]

2. 翻译次数最多的博客文章。若并列，则返回字母顺序靠前的文章。

>>> TranslatedBlogArticle = aliased(BlogArticle)
>>> q = (select(BlogArticle  , func.count(BlogArticle.id))
        .join(TranslatedBlogArticle.translation_of)
        .group_by(BlogArticle)
        .order_by(func.count(BlogArticle.id).desc(), BlogArticle.title)
        .limit(1))
>>> session.scalar(q)
BlogArticle(63, "Business seven ability cup church similar itself")

3. 2022年3月按语言分类的浏览量。

>>> page_views = func.count(BlogView.id).label(None)
>>> q = (select(Language, page_views)
        .join(Language.blog_articles)
        .join(BlogArticle.views)
        .where(BlogView.timestamp.between(
            datetime(2022, 3, 1), datetime(2022, 4, 1)))
        .group_by(Language)
        .order_by(page_views.desc()))
>>> session.execute(q).all()
[(Language(1, "English"), 2155), (Language(3, "French"), 512),
(Language(2, "German"), 417), (Language(6, "Portuguese"), 404),
(Language(5, "Spanish"), 305), (Language(4, "Italian"), 283)]

4. 仅考虑德语内容的文章浏览量。

>>> page_views = func.count(BlogView.id).label(None)
>>> q = (select(BlogArticle, page_views)
        .join(BlogArticle.views)
        .join(BlogArticle.language)
        .where(Language.name == 'German')
        .group_by(BlogArticle)
        .order_by(page_views.desc()))
>>> session.execute(q).all()
[ ... 21 results ... ]

5. 2022年1月至12月每月浏览量。

>>> month = func.extract('month', BlogView.timestamp).label(None)
>>> year = func.extract('year', BlogView.timestamp).label(None)
>>> page_views = func.count(BlogView.id).label(None)
>>> q = (select(year, month, page_views)
        .where(BlogView.timestamp.between(
            datetime(2022, 1, 1), datetime(2023, 1, 1)))
        .group_by(year, month)
        .order_by(year, month))
>>> session.execute(q).all()
[(2022, 1, 3649), (2022, 2, 3287), (2022, 3, 4076), (2022, 4, 3820),
(2022, 5, 4034), (2022, 6, 3659), (2022, 7, 3900), (2022, 8, 3705),
(2022, 9, 3639), (2022, 10, 4066), (2022, 11, 4034), (2022, 12, 3925)]

6. 2022年2月每日浏览量。

>>> day = func.extract('day', BlogView.timestamp).label(None)
>>> month = func.extract('month', BlogView.timestamp).label(None)
>>> year = func.extract('year', BlogView.timestamp).label(None)
>>> page_views = func.count(BlogView.id).label(None)
>>> q = (select(year, month, day, page_views)
        .where(BlogView.timestamp.between(
            datetime(2022, 2, 1), datetime(2022, 3, 1)))
        .group_by(year, month, day)
        .order_by(year, month, day))
>>> session.execute(q).all()
[ ... 28 results ... ]

感谢访问我的博客！如果喜欢这篇文章，请考虑通过 Buy me a coffee 支持我的工作，让我保持咖啡因充足。谢谢！