在python中使用SQLAlchemy查詢(xún)PostgreSQL視圖的流程步驟
作為軟件開(kāi)發(fā)人員，查詢(xún) PostgreSQL 視圖是一項(xiàng)常見(jiàn)任務(wù)。使用視圖（代表SQL 查詢(xún)輸出的虛擬表）被認(rèn)為是處理關(guān)系數(shù)據(jù)庫(kù)時(shí)的有效方法。本文介紹如何在 Python 中使用 SQLAlchemy 查詢(xún) PostgreSQL 視圖。在直接進(jìn)行演示之前，以下是我們將使用的所有工具的概述。
先決條件
- 對(duì) Python 和使用Python 的機(jī)器有很好的了解。
- 了解 SQL、PostgreSQL 和 SQLAlchemy 的基本概念。
- Postgres 安裝在您的本地計(jì)算機(jī)上。
所需模塊
pip install psycopg2 pip install sqlalchemy
使用 SQLAlchemy 查詢(xún) (PostgreSQL) 視圖的步驟
創(chuàng)建數(shù)據(jù)庫(kù)和用戶(hù)
現(xiàn)在，我們首先通過(guò)使用用戶(hù)創(chuàng)建數(shù)據(jù)庫(kù)來(lái)設(shè)置 PostgreSQL，并授予該用戶(hù)對(duì)所創(chuàng)建數(shù)據(jù)庫(kù)的所有必需權(quán)限。
# 創(chuàng)建一個(gè)名為demo的數(shù)據(jù)庫(kù) CREATE DATABASE demo; # 創(chuàng)建了一個(gè)用戶(hù)脫模器，密碼為12345678 CREATE USER demouser WITH PASSWORD '12345678'; # 已將客戶(hù)端編碼配置為utf8 ALTER ROLE demouser SET client_encoding TO 'utf8'; ALTER ROLE demouser SET default_transaction_isolation TO 'read committed'; ALTER ROLE demouser SET timezone TO 'UTC'; # 授予所有必需的權(quán)限，以便在demodb上卸載 GRANT ALL PRIVILEGES ON DATABASE demo TO demouser;
設(shè)置Python開(kāi)發(fā)環(huán)境
到目前為止，我們已經(jīng)創(chuàng)建并配置了一個(gè)數(shù)據(jù)庫(kù)用戶(hù)，現(xiàn)在讓我們配置用于開(kāi)發(fā)的虛擬環(huán)境，這一步可以跳過(guò)，但始終建議為每個(gè)項(xiàng)目使用專(zhuān)用的開(kāi)發(fā)環(huán)境，以避免依賴(lài)沖突，這可以通過(guò)以下方式實(shí)現(xiàn) Python 虛擬環(huán)境。
mkdir gfg cd gfg
文件夾的名稱(chēng)并不重要，您可以將其命名為任何您想要的名稱(chēng)，然后 cd （更改目錄）進(jìn)入新創(chuàng)建的目錄，然后運(yùn)行以下命令，為您的項(xiàng)目創(chuàng)建虛擬環(huán)境。
to create a virtual environment python -m venv venv to activate the virtual environment .\venv\Scripts\activate
使用 Python 連接到 Postgres
在此步驟中，我們將連接到之前創(chuàng)建的“demo”數(shù)據(jù)庫(kù)并創(chuàng)建一個(gè)視圖名稱(chēng) demo_view，然后對(duì)其進(jìn)行查詢(xún)。
使用 Python 連接到 Postgres 可以使用我們剛剛安裝的 psycopg2 適配器來(lái)實(shí)現(xiàn)，如圖所示，
from sqlalchemy import create_engine
engine = create_engine('postgresql://demouser:12345678@localhost:5432/demo')
# 連接到數(shù)據(jù)庫(kù)并打印連接成功（如果已連接）
with engine.connect() as conn:
print("Connection successful")輸出：
創(chuàng)建所需的表
要使用 Python 查詢(xún) PostgreSQL 視圖，我們首先需要數(shù)據(jù)庫(kù)中存在一些數(shù)據(jù)，但目前數(shù)據(jù)庫(kù)中還沒(méi)有
在將一些數(shù)據(jù)插入數(shù)據(jù)庫(kù)之前，我們需要通過(guò)將其定義為 SQLAlchemy 表來(lái)創(chuàng)建視圖，如下所示，
from sqlalchemy import create_engine, MetaData, Table,
Column, Integer, String, DateTime
# 使用SQLAlchemy創(chuàng)建數(shù)據(jù)庫(kù)引擎并連接到數(shù)據(jù)庫(kù)服務(wù)器
engine = create_engine('postgresql:'+
'//demouser:12345678@localhost:5432/demo')
# 創(chuàng)建元數(shù)據(jù)對(duì)象
metadata = MetaData()
# 為演示視圖創(chuàng)建一個(gè)表對(duì)象
demo_view = Table('demo_view', metadata,
Column('id', Integer, primary_key=True),
Column('name', String),
Column('created_at', DateTime)
)
# 在數(shù)據(jù)庫(kù)中創(chuàng)建演示視圖
metadata.create_all(engine)輸出：
將數(shù)據(jù)插入 Postgres
現(xiàn)在我們已經(jīng)配置了所有內(nèi)容并創(chuàng)建了 demo_view 表，讓我們插入一些數(shù)據(jù)并查詢(xún) demo_view 表，該表將使用以下 Python 腳本返回一個(gè)視圖，
# 在演示視圖中插入一些數(shù)據(jù)
with engine.connect() as conn:
conn.execute(demo_view.insert(), [
{'name': 'John', 'created_at': '2021-07-01'},
{'name': 'Jane', 'created_at': '2021-07-02'},
{'name': 'Joe', 'created_at': '2021-07-03'}
])
# commit the changes
conn.commit()
# 查詢(xún)demo視圖以獲取name和createdAt列
query = select().select_from(demo_view).with_only_columns(
demo_view.c.name, demo_view.c.created_at)
# 執(zhí)行查詢(xún)并打印結(jié)果
with engine.connect() as conn:
result = conn.execute(query).fetchall()
for row in result:
print(row)輸出：
上面的 Python 腳本首先包含 2 個(gè)部分，其中我們將一些數(shù)據(jù)插入到 demo_view 表中，然后查詢(xún)同一個(gè)表以返回僅包含 name 和created_at 列的視圖。最后，它將打印從數(shù)據(jù)庫(kù)返回的視圖中的所有行，
如果我們檢查數(shù)據(jù)庫(kù)，數(shù)據(jù)已插入到 demo_view 表中，
使用 SQLAlchemy 查詢(xún) Postgres 視圖
查詢(xún)數(shù)據(jù)庫(kù)視圖是指查找視圖中滿(mǎn)足指定條件的行。以下是查詢(xún)數(shù)據(jù)庫(kù)視圖的不同方法：
使用 Group By 子句查詢(xún)數(shù)據(jù)庫(kù)視圖
Group By 子句根據(jù)所選列對(duì)視圖中的行進(jìn)行分組。以下查詢(xún)顯示 demo_view 中每個(gè)名稱(chēng)的行數(shù)。
query = select().select_from(demo_view).with_only_columns( demo_view.c.name, func.count(demo_view.c.name) ).group_by(demo_view.c.name) # 執(zhí)行查詢(xún)并打印結(jié)果 with engine.connect() as conn: result = conn.execute(query).fetchall() for row in result: print(row)
輸出：
上面的查詢(xún)將按名稱(chēng)返回 demo_view 中的行數(shù)。
使用distinct關(guān)鍵字查詢(xún)數(shù)據(jù)庫(kù)視圖
unique 關(guān)鍵字返回視圖中不同的行。以下查詢(xún)返回 demo_view 中的唯一名稱(chēng)。
query = select().select_from( demo_view).with_only_columns( demo_view.c.name).distinct() # 執(zhí)行查詢(xún)并打印結(jié)果 with engine.connect() as conn: result = conn.execute(query).fetchall() for row in result: print(row)
輸出：
上面的查詢(xún)返回 demo_view 中的唯一名稱(chēng)。
使用 Order By 子句查詢(xún)數(shù)據(jù)庫(kù)視圖
Order By 子句根據(jù)指定的列對(duì)視圖中的行進(jìn)行排序。以下查詢(xún)返回 demo_view 中按created_at 列排序的行。
query = select().select_from( demo_view).order_by( demo_view.c.created_at) # 執(zhí)行查詢(xún)并打印結(jié)果 with engine.connect() as conn: result = conn.execute(query).fetchall() for row in result: print(row)
輸出：
上面的查詢(xún)返回 demo_view 中按created_at 列排序的行。
使用 Sum 函數(shù)查詢(xún)數(shù)據(jù)庫(kù)視圖
Sum 函數(shù)返回指定列中的值的總和。以下查詢(xún)返回 demo_view 的 id 列中的值的總和。
query = select().select_from( demo_view).order_by( demo_view.c.created_at) # 執(zhí)行查詢(xún)并打印結(jié)果 with engine.connect() as conn: result = conn.execute(query).fetchall() for row in result: print(row)
輸出：
上面的查詢(xún)將返回 demo_view 的 id 列中的值的總和。
使用 Avg 函數(shù)查詢(xún)數(shù)據(jù)庫(kù)視圖
Avg 函數(shù)返回指定列中值的平均值。以下查詢(xún)返回 demo_view 中 id 列值的平均值。
query = select().select_from( demo_view).with_only_columns( func.avg(demo_view.c.id)) # 執(zhí)行查詢(xún)并打印結(jié)果 with engine.connect() as conn: result = conn.execute(query).fetchall() for row in result: print(row)
輸出：
上面的查詢(xún)將返回demo_view中id列值的平均值。
使用 Count 函數(shù)查詢(xún)數(shù)據(jù)庫(kù)視圖
Count 函數(shù)返回視圖中特定列或?qū)傩缘挠?jì)數(shù)或行數(shù)。
query = select().select_from( demo_view).with_only_columns( func.count(demo_view.c.id)) # 執(zhí)行查詢(xún)并打印結(jié)果 with engine.connect() as conn: result = conn.execute(query).fetchall() for row in result: print(row)
輸出：
上面的查詢(xún)返回 demo_view 中的行數(shù)。
使用 Min 函數(shù)查詢(xún)數(shù)據(jù)庫(kù)視圖
sqlalchemy 提供的 Min 函數(shù)返回指定列的所有行中特定屬性的最小值或最小值。以下是在 id 列上使用 Min 函數(shù)實(shí)現(xiàn)的查詢(xún)的演示，
query = select().select_from( demo_view).with_only_columns( func.min(demo_view.c.id)) # 執(zhí)行查詢(xún)并打印結(jié)果 with engine.connect() as conn: result = conn.execute(query).fetchall() for row in result: print(row)
輸出：
上面的查詢(xún)將返回 demo_view 中 id 列的最小值。
使用 Max 函數(shù)查詢(xún)數(shù)據(jù)庫(kù)視圖
Max 函數(shù)與 Min 相同但完全相反，返回查詢(xún)中指定列的屬性最大值。以下查詢(xún)返回 demo_view 中 id 列的最大值。
query = select().select_from( demo_view).with_only_columns( func.max(demo_view.c.id)) # 執(zhí)行查詢(xún)并打印結(jié)果 with engine.connect() as conn: result = conn.execute(query).fetchall() for row in result: print(row)
輸出：
上面的查詢(xún)將返回 demo_view 中 id 列的最大值。
使用“in”關(guān)鍵字查詢(xún)數(shù)據(jù)庫(kù)視圖
“in”關(guān)鍵字返回視圖中指定列中的值與指定列表中的任何值匹配的行。以下查詢(xún)返回 demo_view 中名稱(chēng)列值與指定列表中的值匹配的行。
query = select('*').select_from(
demo_view).where(
cast(
demo_view.c.name, String
).in_(['John', 'Jane']))
# 執(zhí)行查詢(xún)并打印結(jié)果
with engine.connect() as conn:
result = conn.execute(query).fetchall()
for row in result:
print(row)輸出：
上面的查詢(xún)返回 demo_view 中名稱(chēng)列值與給定列表中的任何值匹配的行。
使用“and”關(guān)鍵字查詢(xún)數(shù)據(jù)庫(kù)視圖
關(guān)鍵字“and”返回視圖中匹配所有指定條件的行。以下查詢(xún)返回 demo_view 中具有與指定值匹配的 name 列值和與指定值匹配的 id 列值的行。
query = select().select_from( demo_view).where( and_(demo_view.c.name == "John", demo_view.c.id == 1)) # 執(zhí)行查詢(xún)并打印結(jié)果 with engine.connect() as conn: result = conn.execute(query).fetchall() for row in result: print(row)
使用“or”關(guān)鍵字查詢(xún)數(shù)據(jù)庫(kù)視圖
“or”函數(shù)返回視圖中與任何指定條件匹配的行，與邏輯或運(yùn)算符相同。以下查詢(xún)將返回 demo_view 中 name 列值等于“John”或 id 列值等于 2 的行，
query = select('*').select_from(demo_view).where(
(demo_view.c.name == "John") | (demo_view.c.id == 2)
)
# 執(zhí)行查詢(xún)并打印結(jié)果
with engine.connect() as conn:
result = conn.execute(query).fetchall()
for row in result:
print(row)輸出：
使用“not”關(guān)鍵字查詢(xún)數(shù)據(jù)庫(kù)視圖
“not”關(guān)鍵字返回視圖中與指定條件不匹配的行。以下查詢(xún)返回 demo_view 中名稱(chēng)列值不與指定值匹配的行。
query = select('*').select_from(demo_view).where(
not_(demo_view.c.name == "John")
)
# 執(zhí)行查詢(xún)并打印結(jié)果
with engine.connect() as conn:
result = conn.execute(query).fetchall()
for row in result:
print(row)輸出 ：
以上就是在python中使用SQLAlchemy查詢(xún)PostgreSQL視圖的流程步驟的詳細(xì)內(nèi)容，更多關(guān)于python SQLAlchemy查詢(xún)PostgreSQL視圖的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
我們都知道Python內(nèi)置方法input只能輸入一行內(nèi)容,我們今天來(lái)實(shí)現(xiàn)輸入多行文本功能,這篇文章主要給大家介紹了關(guān)于Python多行輸入程序的相關(guān)資料,需要的朋友可以參考下2022-04-04
在本篇文章里小編給大家整理是一篇關(guān)于python解決程序無(wú)法打開(kāi)的相關(guān)文章內(nèi)容，需要的朋友們可以學(xué)習(xí)下。2020-06-06
全方位解析Python讀寫(xiě)JSON數(shù)據(jù)的實(shí)戰(zhàn)指南
JSON作為一種輕量級(jí)的數(shù)據(jù)交換格式,已成為現(xiàn)代編程中數(shù)據(jù)交換的??主流格式??,本文將全面探討Python中JSON數(shù)據(jù)的讀寫(xiě)操作,有需要的小伙伴可以了解下2025-09-09
這篇文章主要介紹了PyTorch筆記之scatter()函數(shù)的使用，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2020-02-02
LyScript實(shí)現(xiàn)Hook改寫(xiě)MessageBox的方法詳解
LyScript可實(shí)現(xiàn)自定義匯編指令的替換功能。用戶(hù)可自行編寫(xiě)匯編指令，將程序中特定的通用函數(shù)進(jìn)行功能改寫(xiě)與轉(zhuǎn)向操作，此功能原理是簡(jiǎn)單的Hook操作。本文將詳細(xì)介紹Hook改寫(xiě)MessageBox的方法，感興趣的可以了解一下2022-09-09
python實(shí)現(xiàn)計(jì)算資源圖標(biāo)crc值的方法
這篇文章主要介紹了python實(shí)現(xiàn)計(jì)算資源圖標(biāo)crc值的方法,通過(guò)解析資源文件找到icon的數(shù)據(jù),從而實(shí)現(xiàn)該功能,需要的朋友可以參考下2014-10-10
這篇文章主要是帶大家寫(xiě)一個(gè)利用Turtle庫(kù)制作的一款海龜闖關(guān)的三大迷宮，文中的示例代碼講解詳細(xì)，對(duì)我們學(xué)習(xí)Python有一定幫助，感興趣的可以了解一下2022-04-04
完美解決keras保存好的model不能成功加載問(wèn)題
這篇文章主要介紹了完美解決keras保存好的model不能成功加載問(wèn)題，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-06-06
python實(shí)現(xiàn)簡(jiǎn)單的名片管理系統(tǒng)
這篇文章主要為大家詳細(xì)介紹了python實(shí)現(xiàn)名片管理系統(tǒng)，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-04-04
解決pip install安裝包成功后仍不可使用的問(wèn)題
文章講述了在使用pip install安裝包后,如果在控制臺(tái)中找不到包,但實(shí)際上包已經(jīng)安裝成功,這是因?yàn)榭刂婆_(tái)使用的Python解釋器與腳本運(yùn)行時(shí)使用的解釋器不同,建議檢查并確保兩者使用同一個(gè)Python解釋器2025-11-11