當我們考慮直接與數(shù)據(jù)庫交互的軟件時，許多人設想了一種類似的管理連接、顯式提交查詢的模式，以及同樣令人厭煩的樣板文件。像PyMySQL 和Psycopg2這樣的 Python 庫完全符合這種范式，并維持了我們所接受的現(xiàn)狀。SQLAlchemy通過抽象軟件中平凡且瑣碎的數(shù)據(jù)庫，已成為 Python 最具標志性的庫之一。

它也是我曾經(jīng)不幸瀏覽過的 Python 庫中最糟糕的文檔的所在地。

SQLAlchemy 的 API 提供了一種更好、更簡單、更快的方式來使用關系數(shù)據(jù)庫。對于那些試圖理解偽裝成SQLAlchemy 文檔的深奧術語的人來說，這毫無用處，這些術語打了你一巴掌，說：“我才不在乎你呢。”

我在這里想說的是，我確實很關心你，并且只想以 SQLAlchemy 教程的形式表達這些感受。

SQL 還是 ORM？

SQLAlchemy 可以實現(xiàn)兩個目的：使 SQL 交互更容易并充當成熟的 ORM。即使您對實現(xiàn) ORM 不感興趣（或者不知道 ORM 是什么），SQLAlchemy 作為簡化 SQL 數(shù)據(jù)庫連接和執(zhí)行原始查詢的工具也是非常寶貴的。我以為這是我寫的原始語句，但事實證明 SQLAlchemy 是這樣描述自己的：

SQLAlchemy 由兩個不同的組件組成，稱為核心和ORM。Core 本身是一個功能齊全的 SQL 抽象工具包，為各種 DBAPI 實現(xiàn)和行為提供平滑的抽象層，以及允許通過生成式 Python 表達式來表達 SQL 語言的 SQL 表達式語言。

連接到數(shù)據(jù)庫

SQLAlchemy 為我們提供了一些管理數(shù)據(jù)庫連接的選項，但它們都始于引擎的概念?！耙妗笔谴頂?shù)據(jù)庫的 Python 對象。通過傳入目標數(shù)據(jù)庫的連接信息來預先創(chuàng)建引擎。

創(chuàng)建引擎后，我們可以隨時通過與引擎對象交互來快速使用我們的數(shù)據(jù)庫。將 SQL 語句直接傳遞給引擎可以讓 SQLAlchemy 處理打開新連接、執(zhí)行并立即關閉連接的操作。對于需要維持大量數(shù)據(jù)傳輸?shù)膹碗s Web 應用程序，最好通過會話保持持續(xù)的數(shù)據(jù)庫連接。我們將在下一篇文章中討論這一點。

創(chuàng)建引擎的語法非常簡單。create_engine()需要一個位置參數(shù)，它是一個表示連接到數(shù)據(jù)庫的連接信息的字符串：

初始化 SQLAlchemy 數(shù)據(jù)庫引擎：

"""創(chuàng)建數(shù)據(jù)庫連接。"""
from sqlalchemy import create_engine


engine = create_engine(
    "mysql+pymysql://user:password@host:3306/database",
)

傳入的字符串create_engine()是一個連接 URI。假設您有一個數(shù)據(jù)庫可供使用，我們將弄清楚您的 URI 應該是什么樣子。

數(shù)據(jù)庫連接 URI

將 URI 分解為幾個部分，如下所示：

數(shù)據(jù)庫連接URI結構：

[DB_TYPE]+[DB_CONNECTOR]://[USERNAME]:[PASSWORD]@[HOST]:[PORT]/[DB_NAME]

• [DB_TYPE]：指定我們要連接的數(shù)據(jù)庫的類型（方言）。SQLAlchemy 可以與所有主流類型的關系數(shù)據(jù)庫接口。根據(jù)您要連接的數(shù)據(jù)庫，替換[DB_TYPE]為匹配的方言：

• MySQL：mysql

• PostgreSQL：postgresql

• SQLite：sqlite

• 甲骨文（呃）：oracle

• Microsoft SQL（稍微不那么惱怒的“呃”）：mssql

• [DB_CONNECTOR]：為了管理您的數(shù)據(jù)庫連接，SQLAlchemy 利用您選擇使用的任何 Python 數(shù)據(jù)庫連接庫。如果您不確定這意味著什么，以下是我針對每種方言推薦的庫：

• MySQL：pymysql、mysqldb

• PostgreSQL：psycopg2、pg8000

• SQLite：（不需要）

• 甲骨文：cx_oracle

• 微軟 SQL：pymssql、pyodbc

pymysql：https://pypi.org/project/PyMySQL/?ref=toymoban.com
mysqldb：https://pypi.org/project/MySQL-python/?ref=toymoban.com
psycopg2：https://pypi.org/project/psycopg2/?ref=toymoban.com
pg8000：https://pypi.org/project/pg8000/?ref=toymoban.com
pymssql：https://pypi.org/project/pymssql/?ref=toymoban.com
pyodbc：https://pypi.org/project/pyodbc/?ref=toymoban.com

之后的變量應該看起來都很熟悉；這些指的是目標數(shù)據(jù)庫的 URL、數(shù)據(jù)庫用戶、該用戶的密碼等。

完成此操作后，我們可以將 SQLAlchemy 與您的連接器一起安裝：

#安裝 SQLAlchemy 和數(shù)據(jù)庫連接器庫
$ pip install sqlalchemy pymysql

附加發(fā)動機配置

除了連接 URI 之外，還可以配置引擎來滿足數(shù)據(jù)庫設置的特定需求（或首選項）。我要求通過 SSL 進行連接；這需要提供一個 PEM 密鑰以提高安全性，我們可以connect_args在創(chuàng)建引擎時通過關鍵字參數(shù)傳遞該密鑰：

#SQLAlchemy 數(shù)據(jù)庫引擎
"""創(chuàng)建數(shù)據(jù)庫連接。"""
from sqlalchemy import create_engine
from config import SQLALCHEMY_DATABASE_PEM


engine = create_engine(
    "mysql+pymysql://user:password@host:3306/database",
    connect_args={"ssl": {"key": SQLALCHEMY_DATABASE_PEM}},
    echo=True,
)

如果您是好奇的類型，您還可以傳遞echo=True給您的引擎，它將在發(fā)生時將正在執(zhí)行的所有SQL 語句打印到您的控制臺（包括連接、表創(chuàng)建等）。這對于了解新來者的情況非常有用，但它很快就會變得煩人和垃圾郵件。

執(zhí)行查詢

我們可以使用 直接在引擎對象上運行即席查詢engine.execute("SELECT * FROM mytable")。當我們調(diào)用execute()引擎時，SQLAlchemy 會處理：

• 打開與我們的數(shù)據(jù)庫的連接。

• 在我們的數(shù)據(jù)庫上執(zhí)行原始 SQL 并返回結果。

• 立即關閉用于運行此查詢的數(shù)據(jù)庫連接以避免掛起連接。

這種與數(shù)據(jù)庫交互的方式稱為顯式無連接執(zhí)行。這種類型的操作是“顯式”的，因為查詢在執(zhí)行時自動提交（與傳統(tǒng)的 Python 庫相比，傳統(tǒng)的 Python 庫遵守PEP-249，除非后面跟著 a ，否則不會運行查詢commit()）。

#PEP-249
https://www.python.org/dev/peps/pep-0249/?ref=toymoban.com#id3

當我們想要按需運行即席查詢時（即：幾乎所有不是面向用戶的應用程序的情況），這非常有用。我們可以獲取或改變我們想要的數(shù)據(jù)，而不必擔心額外的東西。

選擇數(shù)據(jù)

我為此示例設置了一個虛擬數(shù)據(jù)數(shù)據(jù)庫。為了演示使用 SQLAlchemy 有多么容易，我將首先從名為nyc_jobs 的表中獲取一些數(shù)據(jù)：

#執(zhí)行SELECT查詢
"""針對 SQLAlchemy 引擎執(zhí)行原始 SQL 查詢。"""
from typing import List, Optional
import json

from sqlalchemy import text
from sqlalchemy.engine.base import Engine
from sqlalchemy.exc import SQLAlchemyError

from logger import LOGGER


def fetch_job_listings(engine: Engine) -> Optional[List[dict]]:
    """
    Select rows from database and parse as list of dicts.

    :param Engine engine: Database engine to handle raw SQL queries.

    :return: Optional[List[dict]]
    """
    try:
        with engine.begin() as conn:
            result = conn.execute(
                text(
                    "SELECT job_id, agency, business_title, \
                    salary_range_from, salary_range_to \
                    FROM nyc_jobs ORDER BY RAND() LIMIT 10;"
                ),
            )
            results = result.fetchall()
            LOGGER.info(f"Selected rows: {results}"
    except SQLAlchemyError as e:
        LOGGER.error(f"SQLAlchemyError while fetching records: {e}")
    except Exception as e:
        LOGGER.error(f"Unexpected error while fetching records: {e}")

我已經(jīng)傳遞了一個簡單的 SQL 查詢conn.execute()，其中包含我的nyc_jobsSELECT表中的幾列（以隨機順序返回）。讓我們檢查一下輸出：

#查詢返回一個Result對象
<sqlalchemy.engine.result.Result object at 0x10c59bdc0>

天啊，這到底是什么？！什么是Result？這一切是不是好得令人難以置信？

從 SQLAlchemy 1.4.X 開始，該對象ResultProxy已重命名為Result. 檢查您的 SQLAlchemy 版本，了解在獲取結果時要排除哪個對象。

解析查詢結果

Result（是一個有用的數(shù)據(jù)結構，它總結了我們的查詢結果并包裝了結果本身。該Result對象使我們可以輕松地從高層次上了解我們的查詢是如何成功的，方法是讓我們訪問以下屬性rowcount：

#查詢返回的行數(shù)
...


with engine.begin() as conn:
    result = conn.execute(
        text(
            "SELECT job_id, agency, business_title, \
            salary_range_from, salary_range_to \
            FROM nyc_jobs ORDER BY RAND();"
       ),
    )
    results = result.fetchall()
    LOGGER.info(f"Selected {result.rowcount} rows.")

這給了我們...

#輸出results.rowcount
Selected 3123 rows.

...但已經(jīng)夠閑聊了。讓我們通過調(diào)用fetchall()我們的來看看我們的數(shù)據(jù)是什么樣的Result：

#打印從查詢中獲取的行
...


with engine.begin() as conn:
    result = conn.execute(
        text(
            "SELECT job_id, agency, business_title, \
            salary_range_from, salary_range_to \
            FROM nyc_jobs ORDER BY RAND() LIMIT 5;"
       ),
    )
    results = result.fetchall()
    print(f"Selected {result.rowcount} rows.")
    for row in result.fetchall():
        print(row)

現(xiàn)在我們正在談論：

#查詢結果
Selected 5 rows.
(399274, 'NYC HOUSING AUTHORITY', 'Vice President Portfolio Planning, Project Development and Asset Mgt', 86346, 217244)
(399276, 'BOROUGH PRESIDENT-QUEENS', 'Director of Graphic Design', 52524, 81535)
(399276, 'BOROUGH PRESIDENT-QUEENS', 'Director of Graphic Design', 52524, 81535)
(399300, 'NYC HOUSING AUTHORITY', 'VICE PRESIDENT TRANSACTIONS', 86346, 217244)
...

嘿，這些看起來像是包含職位列表的行！

大多數(shù) SQLAlchemy 用戶可能更喜歡將獲取的行格式化為不同的數(shù)據(jù)類型，而不是 SQLAlchemy 對象。一個常見的場景是將行解析為 Python 字典。幸運的是，SQLAlchemy對象有一個用于此目的的Row內(nèi)置方法：_asdict()

#將獲取的數(shù)據(jù)解析為字典列表
...

results_dict = [row._asdict() for row in results]
LOGGER.info(
    f"Selected {result.rowcount} rows: {results_dict}"
)

輸出：

Selected 5 rows: [{'job_id': 312335, 'agency': 'DEPT OF ENVIRONMENT PROTECTION', 'business_title': 'Dynamic CRM Developer', 'salary_range_from': 85823, 'salary_range_to': 121363}, {'job_id': 385305, 'agency': 'POLICE DEPARTMENT', 'business_title': 'Case Management Nurse', 'salary_range_from': 81653, 'salary_range_to': 81653}, {'job_id': 384016, 'agency': 'DEPT OF INFO TECH & TELECOMM', 'business_title': '.NET Developer/Programmer Analyst', 'salary_range_from': 56990, 'salary_range_to': 110000}, {'job_id': 221045, 'agency': 'NYC EMPLOYEES RETIREMENT SYS', 'business_title': 'CERTIFIED IT ADMINISTRATOR (WAN)', 'salary_range_from': 89383, 'salary_range_to': 134914}, {'job_id': 397017, 'agency': 'DEPARTMENT OF BUILDINGS', 'business_title': 'Operations Associate', 'salary_range_from': 31893, 'salary_range_to': 36677}]

非常接近，但有點難以閱讀。使用 Python 的內(nèi)置json庫可以幫助我們：

#將行解析為字典列表
...
import json


results_dict = [row._asdict() for row in results]
LOGGER.info(
    f"Selected {result.rowcount} rows: {json.dumps(results_dict, indent=2)}"
)

現(xiàn)在，無論您想要實現(xiàn)什么目標，您都擁有了一種肯定能滿足您需求的格式：

[
...
  {
    "job_id": 398955,
    "agency": "DEPT OF INFO TECH & TELECOMM",
    "business_title": "Cyber Threat Analyst",
    "salary_range_from": 76500,
    "salary_range_to": 95000
  },
  {
    "job_id": 380613,
    "agency": "DEPT OF INFO TECH & TELECOMM",
    "business_title": "Vulnerability Manager",
    "salary_range_from": 69940,
    "salary_range_to": 165000
  },
  {
    "job_id": 390870,
    "agency": "DEPT OF INFO TECH & TELECOMM",
    "business_title": "Cyber Threat Analyst",
    "salary_range_from": 75000,
    "salary_range_to": 95000
  },
  {
    "job_id": 356355,
    "agency": "DEPT OF ENVIRONMENT PROTECTION",
    "business_title": "Engineering Technician I",
    "salary_range_from": 36239,
    "salary_range_to": 41675
  },
...
]

更新行

我們可以像運行 SELECT 查詢一樣運行 UPDATE 查詢，但需要注意的是，存在錯誤的UPDATE 查詢。讓我們確保避免這些。

我從表中隨機選擇一行進行更新。我將更改第229837行，它的 JSON 形式如下所示：

#SELECT * FROM nyc_jobs WHERE job_id = 229837;
[
 {
   "job_id": 229837,
   "agency": "DEPT OF INFO TECH & TELECOMM",
   "business_title": "Senior Quality Oversight Analyst",
   "job_category": "Information Technology & Telecommunications",
   "salary_range_from": 58675,
   "salary_range_to": 125000,
   "salary_frequency": "Annual",
   "work_location": "75 Park Place New York Ny",
   "division": "General Counsel",
   "created_at": "2016-02-17T00:00:00.000",
   "updated_at": "2016-02-17T00:00:00.000"
 }
]

讓我們編寫一個潛在危險的 SQL 查詢，它會引入一些有問題的字符：

#UPDATE包含引號和表情符號的查詢
UPDATE
    nyc_jobs
SET
    business_title = 'Senior QA Scapegoat ??',
    job_category = 'Info <>!#%%Technology%%#^&%* & Telecom'
WHERE
    job_id = 229837;

如果值包含肯定會中斷操作的運算符，則嘗試更新是一個壞主意。讓我們看看發(fā)生了什么：

#執(zhí)行有問題的 UPDATE 查詢
with engine.begin() as conn:
    result = engine.execute(
        text(
            "UPDATE nyc_jobs SET business_title = 'Senior QA Scapegoat ??', \
            job_category = 'Information? <>!#%%Technology!%%#^&%* & Telecom' \
            WHERE job_id = 229837;"
        )
    )
    LOGGER.info(result.rowcount)

運行這個會惹惱一切：

#錯誤 SQL UPDATE 查詢的堆棧跟蹤
2021-01-09 15:44:14,122 INFO sqlalchemy.engine.base.Engine UPDATE nyc_jobs SET business_title = 'Senior QA Scapegoat ??',         job_category = 'Information <>!#%%Technology%%#^&%* & Telecommunications'         WHERE job_id = 229837;
2021-01-09 15:44:14,123 INFO sqlalchemy.engine.base.Engine {}
2021-01-09 15:44:14,123 INFO sqlalchemy.engine.base.Engine ROLLBACK
Traceback (most recent call last):
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/main.py", line 4, in <module>
    init_script()
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/sqlalchemy_tutorial/__init__.py", line 17, in init_script
    rows_updated = update_job_listing(db)
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/sqlalchemy_tutorial/queries.py", line 18, in update_job_listing
    result = db.execute(
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 2235, in execute
    return connection.execute(statement, *multiparams, **params)
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1003, in execute
    return self._execute_text(object_, multiparams, params)
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1172, in _execute_text
    ret = self._execute_context(
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1316, in _execute_context
    self._handle_dbapi_exception(
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1514, in _handle_dbapi_exception
    util.raise_(exc_info[1], with_traceback=exc_info[2])
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/.venv/lib/python3.8/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
    raise exception
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/.venv/lib/python3.8/site-packages/sqlalchemy/engine/base.py", line 1276, in _execute_context
    self.dialect.do_execute(
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/.venv/lib/python3.8/site-packages/sqlalchemy/engine/default.py", line 609, in do_execute
    cursor.execute(statement, parameters)
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/.venv/lib/python3.8/site-packages/pymysql/cursors.py", line 146, in execute
    query = self.mogrify(query, args)
  File "/Users/toddbirchard/projects/sqlalchemy-tutorial/.venv/lib/python3.8/site-packages/pymysql/cursors.py", line 125, in mogrify
    query = query % self._escape_args(args, conn)
TypeError: * wants int

對我們來說幸運的是，SQLAlchemy 有一個名為 的方法text()，它可以轉義在此類查詢中發(fā)現(xiàn)的危險字符。始終將您的查詢包裝在該方法中：

#將查詢包裝在text()
...
from sqlalchemy import text

with engine.begin() as conn:
    result = conn.execute(
        text(
            "UPDATE nyc_jobs SET business_title = 'Senior QA Scapegoat ??',\
            job_category = 'Information <>!#%%Technology%%#^&%* & Telecom' \
            WHERE job_id = 229837;"
         )
    )
    LOGGER.success(f"Updated {result.rowcount} row: {result}")

輸出：

08-31-2023 08:11:10 | SUCCESS: Updated 1 row: <sqlalchemy.engine.cursor.CursorResult object at 0x107021780>

現(xiàn)在，我們沒有收到任何錯誤，并且返回計數(shù)1以指示更新了單行。

關鍵詞：SQLAlchemy，Python數(shù)據(jù)庫交互，引擎，連接URI，數(shù)據(jù)庫連接文章來源地址http://www.zghlxwxcb.cn/article/581.html

到此這篇關于使用SQLAlchemy簡化Python數(shù)據(jù)庫交互的文章就介紹到這了,更多相關內(nèi)容可以在右上角搜索或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！