沃家組網(wǎng)網(wǎng)絡(luò)服務(wù)費(fèi),寧波seo外包服務(wù)平臺(tái),寧波網(wǎng)站建設(shè)設(shè)計(jì)報(bào)告,永久免費(fèi)素材網(wǎng)站大家好#xff0c;我是jobleap.cn的小九。 psycopg2-binary 是 Python 連接 PostgreSQL 數(shù)據(jù)庫(kù)的核心庫(kù)#xff08;psycopg2 的預(yù)編譯二進(jìn)制版本#xff0c;無(wú)需編譯依賴(lài)#xff0c;開(kāi)箱即用#xff09;#xff0c;本文將從環(huán)境準(zhǔn)備、核心 API 講解到實(shí)戰(zhàn)案例#xff0c…大家好我是jobleap.cn的小九。psycopg2-binary 是 Python 連接 PostgreSQL 數(shù)據(jù)庫(kù)的核心庫(kù)psycopg2的預(yù)編譯二進(jìn)制版本無(wú)需編譯依賴(lài)開(kāi)箱即用本文將從環(huán)境準(zhǔn)備、核心 API 講解到實(shí)戰(zhàn)案例全面串聯(lián)其常用用法幫助你掌握 PostgreSQL 數(shù)據(jù)庫(kù)的 Python 操作全流程。一、環(huán)境準(zhǔn)備1. 安裝 psycopg2-binary使用 pip 快速安裝推薦 Python 3.6 版本# 安裝最新版pip install psycopg2-binary# 安裝指定版本如適配特定 PostgreSQL 版本pip install psycopg2-binary2.9.92. 前置條件已安裝并啟動(dòng) PostgreSQL 服務(wù)本地/遠(yuǎn)程擁有可訪問(wèn)的 PostgreSQL 數(shù)據(jù)庫(kù)、用戶(hù)名和密碼確保目標(biāo)數(shù)據(jù)庫(kù)端口默認(rèn) 5432未被防火墻攔截。二、核心概念與基礎(chǔ) APIpsycopg2 的核心操作圍繞連接Connection和游標(biāo)Cursor展開(kāi)Connection負(fù)責(zé)與 PostgreSQL 數(shù)據(jù)庫(kù)建立網(wǎng)絡(luò)連接管理事務(wù)Cursor基于連接創(chuàng)建的操作句柄用于執(zhí)行 SQL 語(yǔ)句、獲取查詢(xún)結(jié)果。1. 數(shù)據(jù)庫(kù)連接connect()psycopg2.connect()是創(chuàng)建數(shù)據(jù)庫(kù)連接的核心函數(shù)支持通過(guò)參數(shù)或 DSN 字符串傳參常用參數(shù)如下參數(shù)說(shuō)明默認(rèn)值host數(shù)據(jù)庫(kù)服務(wù)器地址localhostport數(shù)據(jù)庫(kù)端口5432dbname/database目標(biāo)數(shù)據(jù)庫(kù)名-user數(shù)據(jù)庫(kù)用戶(hù)名當(dāng)前系統(tǒng)用戶(hù)password數(shù)據(jù)庫(kù)密碼-sslmodeSSL 連接模式如 requiredisable基礎(chǔ)連接示例importpsycopg2frompsycopg2importOperationalErrordefcreate_connection(db_name,db_user,db_password,db_host,db_port):創(chuàng)建數(shù)據(jù)庫(kù)連接并返回 Connection 對(duì)象connectionNonetry:connectionpsycopg2.connect(databasedb_name,userdb_user,passworddb_password,hostdb_host,portdb_port,)print(PostgreSQL 連接成功 ?)exceptOperationalErrorase:print(f連接失敗 ?{e})returnconnection# 替換為你的數(shù)據(jù)庫(kù)信息conncreate_connection(db_nametest_db,db_userpostgres,db_password123456,db_hostlocalhost,db_port5432)2. 游標(biāo)創(chuàng)建與 SQL 執(zhí)行cursor()/execute()創(chuàng)建連接后需通過(guò)conn.cursor()創(chuàng)建游標(biāo)再用游標(biāo)執(zhí)行 SQL 語(yǔ)句cursor.execute(sql, params)執(zhí)行單條 SQL 語(yǔ)句支持參數(shù)化查詢(xún)cursor.executemany(sql, params_list)批量執(zhí)行相同結(jié)構(gòu)的 SQL 語(yǔ)句psycopg2.extras.execute_batch(cursor, sql, params_list, page_size100)高性能批量執(zhí)行推薦替代executemany。1創(chuàng)建數(shù)據(jù)表示例defcreate_table(connection):創(chuàng)建用戶(hù)表userscreate_table_query CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, name VARCHAR(50) NOT NULL, age INT, email VARCHAR(100) UNIQUE, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); try:# 創(chuàng)建游標(biāo)cursorconnection.cursor()# 執(zhí)行 SQLcursor.execute(create_table_query)# 提交事務(wù)psycopg2 默認(rèn)關(guān)閉自動(dòng)提交必須手動(dòng)提交connection.commit()print(數(shù)據(jù)表創(chuàng)建成功 ?)exceptExceptionase:print(f創(chuàng)建表失敗 ?{e})# 異常時(shí)回滾事務(wù)connection.rollback()finally:# 關(guān)閉游標(biāo)避免資源泄漏cursor.close()# 調(diào)用創(chuàng)建表函數(shù)ifconn:create_table(conn)2參數(shù)化查詢(xún)防 SQL 注入關(guān)鍵psycopg2 使用%s作為占位符而非 Python 的{}或%參數(shù)需以元組/列表傳入。definsert_single_user(connection,name,age,email):插入單條用戶(hù)數(shù)據(jù)參數(shù)化查詢(xún)insert_query INSERT INTO users (name, age, email) VALUES (%s, %s, %s) ON CONFLICT (email) DO NOTHING; # 避免重復(fù)插入 try:cursorconnection.cursor()# 傳入?yún)?shù)元組形式cursor.execute(insert_query,(name,age,email))connection.commit()print(f插入用戶(hù){name}成功 ?)exceptExceptionase:print(f插入失敗 ?{e})connection.rollback()finally:cursor.close()# 插入單條數(shù)據(jù)ifconn:insert_single_user(conn,張三,25,zhangsanexample.com)3批量插入數(shù)據(jù)frompsycopg2importextrasdefbatch_insert_users(connection,users_list):批量插入用戶(hù)數(shù)據(jù)高性能版insert_query INSERT INTO users (name, age, email) VALUES (%s, %s, %s) ON CONFLICT (email) DO NOTHING; try:cursorconnection.cursor()# 高性能批量執(zhí)行page_size 控制每次批量提交的條數(shù)extras.execute_batch(cursor,insert_query,users_list,page_size100)connection.commit()print(f批量插入{len(users_list)}條數(shù)據(jù)成功 ?)exceptExceptionase:print(f批量插入失敗 ?{e})connection.rollback()finally:cursor.close()# 批量插入示例數(shù)據(jù)ifconn:users_data[(李四,28,lisiexample.com),(王五,30,wangwuexample.com),(趙六,22,zhaoliuexample.com)]batch_insert_users(conn,users_data)3. 數(shù)據(jù)查詢(xún)fetchone()/fetchmany()/fetchall()執(zhí)行查詢(xún)類(lèi) SQLSELECT后需通過(guò)游標(biāo)獲取結(jié)果cursor.fetchone()獲取下一條結(jié)果返回元組無(wú)數(shù)據(jù)時(shí)返回 Nonecursor.fetchmany(size)獲取指定條數(shù)的結(jié)果返回列表元素為元組cursor.fetchall()獲取所有剩余結(jié)果返回列表元素為元組cursor.rowcount返回受上一條 SQL 影響的行數(shù)查詢(xún)時(shí)為匹配的行數(shù)。查詢(xún)示例defquery_users(connection,age_min0):查詢(xún)年齡大于等于 age_min 的用戶(hù)query SELECT id, name, age, email, create_time FROM users WHERE age %s; try:cursorconnection.cursor()cursor.execute(query,(age_min,))# 方式1獲取單條數(shù)據(jù)# single_user cursor.fetchone()# if single_user:# print(單條結(jié)果, single_user)# 方式2獲取指定條數(shù)如2條# partial_users cursor.fetchmany(2)# print(部分結(jié)果, partial_users)# 方式3獲取所有結(jié)果all_userscursor.fetchall()print(f 查詢(xún)到{cursor.rowcount}條符合條件的用戶(hù))foruserinall_users:# 解析元組id, name, age, email, create_timeprint(fID:{user[0]}, 姓名:{user[1]}, 年齡:{user[2]}, 郵箱:{user[3]}, 創(chuàng)建時(shí)間:{user[4]})exceptExceptionase:print(f查詢(xún)失敗 ?{e})finally:cursor.close()# 查詢(xún)年齡≥25的用戶(hù)ifconn:query_users(conn,age_min25)4. 數(shù)據(jù)更新與刪除更新/刪除操作與插入邏輯一致需注意事務(wù)提交和參數(shù)化defupdate_user_age(connection,email,new_age):根據(jù)郵箱更新用戶(hù)年齡update_query UPDATE users SET age %s WHERE email %s; try:cursorconnection.cursor()cursor.execute(update_query,(new_age,email))connection.commit()ifcursor.rowcount0:print(f更新{email}的年齡為{new_age}成功 ?)else:print(f未找到郵箱為{email}的用戶(hù) ?)exceptExceptionase:print(f更新失敗 ?{e})connection.rollback()finally:cursor.close()defdelete_user(connection,user_id):根據(jù)ID刪除用戶(hù)delete_queryDELETE FROM users WHERE id %s;try:cursorconnection.cursor()cursor.execute(delete_query,(user_id,))connection.commit()ifcursor.rowcount0:print(f刪除ID為{user_id}的用戶(hù)成功 ?)else:print(f未找到ID為{user_id}的用戶(hù) ?)exceptExceptionase:print(f刪除失敗 ?{e})connection.rollback()finally:cursor.close()# 執(zhí)行更新和刪除ifconn:update_user_age(conn,zhangsanexample.com,26)delete_user(conn,3)# 刪除ID為3的用戶(hù)query_users(conn)# 重新查詢(xún)驗(yàn)證結(jié)果5. 事務(wù)管理commit()/rollback()psycopg2 默認(rèn)關(guān)閉「自動(dòng)提交」模式所有修改類(lèi)操作INSERT/UPDATE/DELETE/CREATE都需要手動(dòng)調(diào)用conn.commit()確認(rèn)若執(zhí)行過(guò)程中出現(xiàn)異常需調(diào)用conn.rollback()回滾事務(wù)避免數(shù)據(jù)不一致。事務(wù)回滾示例deftest_transaction(connection):測(cè)試事務(wù)回滾try:cursorconnection.cursor()# 第一步插入數(shù)據(jù)cursor.execute(INSERT INTO users (name, age, email) VALUES (%s, %s, %s),(測(cè)試用戶(hù),99,testexample.com))# 第二步故意觸發(fā)錯(cuò)誤比如插入重復(fù)郵箱cursor.execute(INSERT INTO users (name, age, email) VALUES (%s, %s, %s),(重復(fù)用戶(hù),88,zhangsanexample.com))# 無(wú)異常則提交connection.commit()exceptExceptionase:print(f事務(wù)執(zhí)行失敗觸發(fā)回滾 ?{e})connection.rollback()# 回滾所有未提交的操作finally:cursor.close()# 測(cè)試事務(wù)回滾最終 測(cè)試用戶(hù) 不會(huì)被插入ifconn:test_transaction(conn)query_users(conn)6. 類(lèi)型轉(zhuǎn)換PostgreSQL ? Pythonpsycopg2 會(huì)自動(dòng)完成 PostgreSQL 類(lèi)型與 Python 類(lèi)型的轉(zhuǎn)換常用映射關(guān)系如下PostgreSQL 類(lèi)型Python 類(lèi)型INT/SERIALintVARCHAR/TEXTstrTIMESTAMP/DATEdatetime.datetime/dateBOOLEANboolARRAYlistJSON/JSONBdict/list需導(dǎo)入 extrasJSON 類(lèi)型操作示例frompsycopg2.extrasimportJsondeftest_json_type(connection):測(cè)試 JSON 類(lèi)型字段操作# 1. 先添加 JSON 字段alter_queryALTER TABLE users ADD COLUMN IF NOT EXISTS info JSONB;# 2. 更新 JSON 數(shù)據(jù)update_queryUPDATE users SET info %s WHERE email %s;try:cursorconnection.cursor()cursor.execute(alter_query)# 傳入 Python 字典通過(guò) Json 封裝user_info{hobby:[籃球,編程],address:北京市}cursor.execute(update_query,(Json(user_info),zhangsanexample.com))connection.commit()# 3. 查詢(xún) JSON 字段cursor.execute(SELECT name, info FROM users WHERE email %s;,(zhangsanexample.com,))resultcursor.fetchone()print(f JSON 字段查詢(xún)結(jié)果)print(f姓名{result[0]}, 信息{result[1]})print(f提取 hobby{result[1][hobby]})# 直接按字典訪問(wèn)exceptExceptionase:print(fJSON 操作失敗 ?{e})connection.rollback()finally:cursor.close()ifconn:test_json_type(conn)7. 連接池生產(chǎn)環(huán)境必備頻繁創(chuàng)建/關(guān)閉連接會(huì)消耗大量資源生產(chǎn)環(huán)境建議使用連接池psycopg2.pool復(fù)用連接frompsycopg2importpool# 創(chuàng)建連接池最小1個(gè)最大5個(gè)連接connection_poolpool.SimpleConnectionPool(minconn1,maxconn5,databasetest_db,userpostgres,password123456,hostlocalhost,port5432)defuse_pooled_connection():使用連接池獲取連接# 從池獲取連接connconnection_pool.getconn()ifconn:print( 從連接池獲取連接成功 ?)query_users(conn)# 歸還連接到池不是關(guān)閉connection_pool.putconn(conn)# 測(cè)試連接池use_pooled_connection()# 關(guān)閉連接池程序退出時(shí)connection_pool.closeall()8. 資源釋放使用完連接和游標(biāo)后必須關(guān)閉以釋放資源推薦通過(guò)finally塊確保執(zhí)行# 最終關(guān)閉連接非連接池場(chǎng)景ifconn:try:conn.close()print( 數(shù)據(jù)庫(kù)連接已關(guān)閉 ?)exceptExceptionase:print(f關(guān)閉連接失敗 ?{e})三、完整實(shí)戰(zhàn)腳本串聯(lián)所有 API以下腳本整合了上述所有常用操作可直接運(yùn)行需替換數(shù)據(jù)庫(kù)信息importpsycopg2frompsycopg2importOperationalError,extrasfrompsycopg2.extrasimportJsonfrompsycopg2importpool# 1. 創(chuàng)建數(shù)據(jù)庫(kù)連接或連接池defcreate_connection(db_name,db_user,db_password,db_host,db_port):connectionNonetry:connectionpsycopg2.connect(databasedb_name,userdb_user,passworddb_password,hostdb_host,portdb_port,)print(PostgreSQL 連接成功 ?)exceptOperationalErrorase:print(f連接失敗 ?{e})returnconnection# 2. 創(chuàng)建數(shù)據(jù)表defcreate_table(connection):create_table_query CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, name VARCHAR(50) NOT NULL, age INT, email VARCHAR(100) UNIQUE, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, info JSONB ); try:cursorconnection.cursor()cursor.execute(create_table_query)connection.commit()print(數(shù)據(jù)表創(chuàng)建成功 ?)exceptExceptionase:print(f創(chuàng)建表失敗 ?{e})connection.rollback()finally:cursor.close()# 3. 插入/更新/刪除/查詢(xún)definsert_user(connection,user_data):insert_queryINSERT INTO users (name, age, email) VALUES (%s, %s, %s) ON CONFLICT (email) DO NOTHING;try:cursorconnection.cursor()extras.execute_batch(cursor,insert_query,user_data,page_size100)connection.commit()print(f批量插入{len(user_data)}條數(shù)據(jù)成功 ?)exceptExceptionase:print(f插入失敗 ?{e})connection.rollback()finally:cursor.close()defupdate_user_info(connection,email,info):update_queryUPDATE users SET info %s WHERE email %s;try:cursorconnection.cursor()cursor.execute(update_query,(Json(info),email))connection.commit()print(f更新{email}的擴(kuò)展信息成功 ?)exceptExceptionase:print(f更新失敗 ?{e})connection.rollback()finally:cursor.close()defquery_users(connection,age_min0):querySELECT id, name, age, email, info FROM users WHERE age %s;try:cursorconnection.cursor()cursor.execute(query,(age_min,))all_userscursor.fetchall()print(f 查詢(xún)到{cursor.rowcount}條用戶(hù)數(shù)據(jù))foruserinall_users:print(fID:{user[0]}, 姓名:{user[1]}, 年齡:{user[2]}, 郵箱:{user[3]}, 擴(kuò)展信息:{user[4]})exceptExceptionase:print(f查詢(xún)失敗 ?{e})finally:cursor.close()defdelete_user(connection,user_id):delete_queryDELETE FROM users WHERE id %s;try:cursorconnection.cursor()cursor.execute(delete_query,(user_id,))connection.commit()print(f刪除ID為{user_id}的用戶(hù){成功ifcursor.rowcount0else失敗}?)exceptExceptionase:print(f刪除失敗 ?{e})connection.rollback()finally:cursor.close()# 主流程if__name____main__:# 替換為你的數(shù)據(jù)庫(kù)信息DB_CONFIG{db_name:test_db,db_user:postgres,db_password:123456,db_host:localhost,db_port:5432}# 創(chuàng)建連接conncreate_connection(**DB_CONFIG)ifnotconn:exit(1)# 執(zhí)行核心操作create_table(conn)insert_user(conn,[(張三,25,zhangsanexample.com),(李四,28,lisiexample.com),(王五,30,wangwuexample.com)])update_user_info(conn,zhangsanexample.com,{hobby:[籃球,編程],address:北京市})query_users(conn,age_min25)delete_user(conn,3)query_users(conn,age_min25)# 關(guān)閉連接ifconn:conn.close()print( 數(shù)據(jù)庫(kù)連接已關(guān)閉 ?)四、常見(jiàn)問(wèn)題與注意事項(xiàng)SQL 注入風(fēng)險(xiǎn)嚴(yán)禁拼接 SQL 字符串必須使用%s占位符傳參編碼問(wèn)題PostgreSQL 默認(rèn)編碼為 UTF8Python 腳本需確保編碼一致連接超時(shí)遠(yuǎn)程連接時(shí)需設(shè)置connect_timeout參數(shù)如connect(..., connect_timeout10)大結(jié)果集處理避免使用fetchall()改用fetchone()或fetchmany()分批讀取防止內(nèi)存溢出版本兼容psycopg2-binary 版本需與 PostgreSQL 服務(wù)版本適配如 2.9.x 適配 PostgreSQL 12。五、總結(jié)psycopg2-binary 的核心流程可總結(jié)為創(chuàng)建連接 → 創(chuàng)建游標(biāo) → 執(zhí)行 SQL → 處理結(jié)果 → 提交/回滾事務(wù) → 釋放資源掌握connect()、cursor()、execute()、commit()、fetch*()等核心 API結(jié)合參數(shù)化查詢(xún)、事務(wù)管理和連接池即可安全、高效地實(shí)現(xiàn) PostgreSQL 數(shù)據(jù)庫(kù)的增刪改查。生產(chǎn)環(huán)境中還需注意異常捕獲、資源釋放和性能優(yōu)化如批量操作、索引設(shè)計(jì)確保系統(tǒng)穩(wěn)定運(yùn)行。