企业官网建设流程全解析

1. 三种数据库连接踩坑记：Python 连 Postgres/Mysql/Mongo 到底难在哪

刚入行那会儿，我以为 Python 连数据库就是import一个包、connect()一下、execute()一把梭。结果第一次同时对接 Postgres、Mysql、Mongo 三个库时，光驱动就装错了两次，连接串拼错三次，游标忘了关导致连接池爆掉一次。这篇文章就把这三种数据库的最小可运行示例摊开讲清楚，从驱动安装、连接串拼接，到增删改查验证，每一步都给可复制的代码。

你可能会问，这三种数据库到底适合谁？简单说：Postgres 适合需要事务、复杂查询、JSON 字段的关系型场景；Mysql 是国内中小项目最常用的关系型库，生态成熟；Mongo 适合文档结构灵活、字段经常变的场景，比如日志、爬虫结果、配置中心。三种都会用，基本能覆盖 80% 的后端数据存储需求。

本文的核心检索词就是Python 连接 Postgres/Mysql/Mongo 数据库基本操作。我会先讲清楚每种库的驱动选择和连接串格式，再给出插入加查询的完整验证代码，最后演示怎么把 API 调用统一到 TaoToken 的 Key 通道，让数据库操作和模型调用共用一套凭证管理。实测下来，这套流程跑通一次之后，后面换库换表基本就是改连接串的事。

先说清楚环境前提：Python 3.8+，本地或容器里已经跑着三个数据库实例。如果你还没装数据库，可以用 Docker 快速起三个容器，命令我会在对应章节给出。驱动方面，Postgres 用psycopg2-binary，Mysql 用PyMySQL（比 MySQLdb 更好装，纯 Python 实现），Mongo 用pymongo。这三个包都能用 pip 直接装，不需要编译。

pip install psycopg2-binary PyMySQL pymongo requests

装完之后，你可以用pip list | grep -E "psycopg2|PyMySQL|pymongo"确认一下。如果 psycopg2-binary 装不上，大概率是系统缺 libpq 开发库，Ubuntu 下apt install libpq-dev即可。Mysql 的 MySQLdb 在 Python 3 下经常编译失败，所以我直接推荐 PyMySQL，API 几乎一样，省心。

接下来我会按 Postgres、Mysql、Mongo 的顺序，每种库给出连接、建表、插入、查询的完整代码。然后单独用一节讲 TaoToken 的 Key 通道怎么统一管理 API 调用。最后是常见报错排查，包括 401、连接超时、游标未关闭这些问题。你可以跟着一步步敲，也可以直接复制代码改连接串。

2. TaoToken 前置准备：统一 Key 通道与 API 调用配置

在讲数据库操作之前，先花点时间把 TaoToken 的 Key 通道配好。为什么要在数据库教程里讲这个？因为实际项目里，数据库操作往往和模型调用混在一起，比如你用 Python 从 Postgres 读一批数据，然后调模型做摘要，再把结果写回 Mongo。如果数据库连接和 API 调用各管各的凭证，维护起来很乱。TaoToken 提供统一的 Key 通道，让 API 调用和数据库操作共用一套配置管理思路。

TaoToken 是什么？简单说，它是一个 API 聚合通道，你申请一个 Key，就能通过统一的 Base URL 调用多种模型。官网是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 地址是 https://taotoken.net/api 。注意 API 地址不带 UTM 参数，直接写就行。

前置准备分三步：注册账号、创建 API Key、配置环境变量。注册就不展开讲了，重点说 Key 的创建和配置。登录后进控制台，找到 API Keys 页面，点创建，复制生成的 Key。这个 Key 只显示一次，记得存好。控制台地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，API Keys 页面是 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。

拿到 Key 之后，不要硬编码在代码里。我习惯用环境变量，Linux/Mac 下在~/.bashrc或~/.zshrc里加一行：

export TAOTOKEN_API_KEY="sk-你的实际Key"

Windows 下用系统环境变量设置，或者用.env文件配合python-dotenv。这样代码里用os.getenv("TAOTOKEN_API_KEY")读取，换 Key 不用改代码。

配置好之后，写一个最小的验证脚本，确认 Key 通道能通。这里用requests直接调 API，不依赖任何 SDK：

import os import requests api_key = os.getenv("TAOTOKEN_API_KEY") base_url = "https://taotoken.net/api" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } payload = { "model": "gpt-4o-mini", "messages": [ {"role": "user", "content": "用一句话说明数据库连接池的作用"} ] } resp = requests.post(f"{base_url}/v1/chat/completions", headers=headers, json=payload, timeout=30) print(resp.status_code) print(resp.json()["choices"][0]["message"]["content"])

跑通这个脚本，说明 Key 通道没问题。如果返回 401，检查 Key 是否复制完整、环境变量是否生效。如果返回连接超时，检查网络和 Base URL 是否写对。这一步过了，后面数据库操作和 API 调用就能共用这套配置。

为什么强调统一 Key 通道？因为实际项目里，你可能需要根据数据库查询结果动态调模型，比如从 Mysql 读用户评论，调模型做情感分析，再写回 Postgres。如果 API 调用散落在各处，Key 轮换时改起来很痛苦。统一到 TaoToken 之后，只需要改一个环境变量。

另外，TaoToken 的模型对话页面可以快速测试不同模型的效果，地址是 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 。如果你要做长期编码或 Agent 类任务，可以看 Coding Plan，地址是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。

配置部分就到这里。接下来进入正题，分别讲三种数据库的连接和操作。

3. 可复制配置：Postgres/Mysql/Mongo 连接串与增删改查代码

这一节是全文的核心，每种数据库我都给出完整的连接配置和增删改查代码。你可以直接复制，改连接参数就能跑。为了统一管理，我建议把连接配置写成一个config.py或.env文件，代码里读取。

先给一个统一的配置结构，用字典存三种库的连接参数：

# config.py import os DB_CONFIG = { "postgres": { "host": os.getenv("PG_HOST", "127.0.0.1"), "port": int(os.getenv("PG_PORT", 5432)), "database": os.getenv("PG_DB", "testdb"), "user": os.getenv("PG_USER", "postgres"), "password": os.getenv("PG_PASSWORD", "postgres"), }, "mysql": { "host": os.getenv("MYSQL_HOST", "127.0.0.1"), "port": int(os.getenv("MYSQL_PORT", 3306)), "database": os.getenv("MYSQL_DB", "testdb"), "user": os.getenv("MYSQL_USER", "root"), "password": os.getenv("MYSQL_PASSWORD", "root"), "charset": "utf8mb4", }, "mongo": { "uri": os.getenv("MONGO_URI", "mongodb://127.0.0.1:27017/"), "database": os.getenv("MONGO_DB", "testdb"), }, }

如果你用 Docker 起数据库，命令如下。Postgres：

docker run -d --name pg-test -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=testdb -p 5432:5432 postgres:16

Mysql：

docker run -d --name mysql-test -e MYSQL_ROOT_PASSWORD=root -e MYSQL_DATABASE=testdb -p 3306:3306 mysql:8

Mongo：

docker run -d --name mongo-test -p 27017:27017 mongo:7

三个容器起来之后，用docker ps确认端口映射正确。接下来分别写代码。

3.1 Postgres 连接与增删改查

Postgres 用psycopg2，连接串可以用关键字参数，也可以用 DSN 字符串。我习惯用关键字参数，清晰。建表、插入、查询、更新、删除的完整代码如下：

import psycopg2 from config import DB_CONFIG def pg_demo(): cfg = DB_CONFIG["postgres"] conn = psycopg2.connect( host=cfg["host"], port=cfg["port"], database=cfg["database"], user=cfg["user"], password=cfg["password"], ) cursor = conn.cursor() # 建表 cursor.execute(""" CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, name TEXT NOT NULL, age INT, created_at TIMESTAMP DEFAULT NOW() ) """) conn.commit() # 插入 cursor.execute( "INSERT INTO users (name, age) VALUES (%s, %s) RETURNING id", ("Alice", 30) ) new_id = cursor.fetchone()[0] conn.commit() print(f"插入成功，id={new_id}") # 查询 cursor.execute("SELECT id, name, age FROM users WHERE age > %s", (18,)) rows = cursor.fetchall() for row in rows: print(row) # 更新 cursor.execute("UPDATE users SET age = %s WHERE id = %s", (31, new_id)) conn.commit() # 删除 cursor.execute("DELETE FROM users WHERE id = %s", (new_id,)) conn.commit() cursor.close() conn.close() if __name__ == "__main__": pg_demo()

注意%s占位符，psycopg2 用这个，不是?。参数化查询能防 SQL 注入，别用字符串拼接。RETURNING id是 Postgres 的特性，插入后直接拿自增 ID，很方便。

3.2 Mysql 连接与增删改查

Mysql 用PyMySQL，连接参数和 psycopg2 类似，但占位符是%s也一样。代码如下：

import pymysql from config import DB_CONFIG def mysql_demo(): cfg = DB_CONFIG["mysql"] conn = pymysql.connect( host=cfg["host"], port=cfg["port"], database=cfg["database"], user=cfg["user"], password=cfg["password"], charset=cfg["charset"], cursorclass=pymysql.cursors.DictCursor, ) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS users ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(64) NOT NULL, age INT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) conn.commit() cursor.execute( "INSERT INTO users (name, age) VALUES (%s, %s)", ("Bob", 25) ) conn.commit() new_id = cursor.lastrowid print(f"插入成功，id={new_id}") cursor.execute("SELECT id, name, age FROM users WHERE age > %s", (18,)) rows = cursor.fetchall() for row in rows: print(row) cursor.execute("UPDATE users SET age = %s WHERE id = %s", (26, new_id)) conn.commit() cursor.execute("DELETE FROM users WHERE id = %s", (new_id,)) conn.commit() cursor.close() conn.close() if __name__ == "__main__": mysql_demo()

cursorclass=pymysql.cursors.DictCursor让查询结果返回字典而不是元组，字段名直接可读，推荐加上。lastrowid拿自增 ID。Mysql 建表记得指定ENGINE=InnoDB和utf8mb4，不然中文可能乱码。

3.3 Mongo 连接与增删改查

Mongo 用pymongo，操作方式和关系型库差别较大。Mongo 不需要建表，插入时自动创建集合。代码如下：

from pymongo import MongoClient from config import DB_CONFIG def mongo_demo(): cfg = DB_CONFIG["mongo"] client = MongoClient(cfg["uri"]) db = client[cfg["database"]] col = db["users"] # 插入一条 result = col.insert_one({"name": "Carol", "age": 28, "tags": ["vip", "new"]}) print(f"插入成功，_id={result.inserted_id}") # 插入多条 col.insert_many([ {"name": "Dave", "age": 35}, {"name": "Eve", "age": 22}, ]) # 查询 for doc in col.find({"age": {"$gt": 18}}): print(doc) # 更新 col.update_one({"name": "Carol"}, {"$set": {"age": 29}}) # 删除 col.delete_one({"name": "Carol"}) client.close() if __name__ == "__main__": mongo_demo()

Mongo 的查询条件用字典，$gt是大于，$set是更新字段。insert_one返回的inserted_id是 ObjectId，不是自增整数。Mongo 不需要 commit，写操作默认立即生效。

三种库的代码都跑一遍，确认插入和查询都成功。如果某一步报错，先看下一节的排查清单。

4. 验证请求与成功结果：一次插入加一次查询确认连通

代码写完了，怎么确认真的连通了？我习惯用「一次插入加一次查询」的方式验证，简单直接。下面给一个统一的验证脚本，依次跑三种库，每种库插入一条记录，再查出来打印。

import psycopg2 import pymysql from pymongo import MongoClient from config import DB_CONFIG def verify_postgres(): cfg = DB_CONFIG["postgres"] conn = psycopg2.connect(**{k: cfg[k] for k in ["host", "port", "database", "user", "password"]}) cur = conn.cursor() cur.execute("CREATE TABLE IF NOT EXISTS verify (id SERIAL PRIMARY KEY, msg TEXT)") cur.execute("INSERT INTO verify (msg) VALUES (%s) RETURNING id", ("pg-ok",)) rid = cur.fetchone()[0] conn.commit() cur.execute("SELECT msg FROM verify WHERE id = %s", (rid,)) print("Postgres:", cur.fetchone()[0]) cur.close() conn.close() def verify_mysql(): cfg = DB_CONFIG["mysql"] conn = pymysql.connect(host=cfg["host"], port=cfg["port"], database=cfg["database"], user=cfg["user"], password=cfg["password"], charset=cfg["charset"]) cur = conn.cursor() cur.execute("CREATE TABLE IF NOT EXISTS verify (id INT AUTO_INCREMENT PRIMARY KEY, msg VARCHAR(64))") cur.execute("INSERT INTO verify (msg) VALUES (%s)", ("mysql-ok",)) conn.commit() rid = cur.lastrowid cur.execute("SELECT msg FROM verify WHERE id = %s", (rid,)) print("Mysql:", cur.fetchone()[0]) cur.close() conn.close() def verify_mongo(): cfg = DB_CONFIG["mongo"] client = MongoClient(cfg["uri"]) col = client[cfg["database"]]["verify"] rid = col.insert_one({"msg": "mongo-ok"}).inserted_id doc = col.find_one({"_id": rid}) print("Mongo:", doc["msg"]) client.close() if __name__ == "__main__": verify_postgres() verify_mysql() verify_mongo()

跑这个脚本，预期输出：

Postgres: pg-ok Mysql: mysql-ok Mongo: mongo-ok

三行都打印出来，说明三种数据库都连通了。如果某一行报错，对照下一节的排查清单。这个验证脚本的好处是，每种库都走了一遍「建表、插入、查询」的完整链路，比单纯SELECT 1更能暴露问题。

验证通过之后，你可以把 TaoToken 的 API 调用也加进来，比如从 Postgres 读一条数据，调模型处理，再写回 Mongo。这样就把数据库操作和 API 调用串起来了。TaoToken 的模型对话页面可以快速测试模型，地址是 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 。

5. 本篇常见错排查：401、连接超时、游标未关闭怎么解

这一节汇总我在实际项目里踩过的坑，按报错信息分类，给出排查步骤。

报错一：psycopg2.OperationalError: connection to server at "127.0.0.1", port 5432 failed: Connection refused

这是 Postgres 连接被拒。先确认容器或服务是否在跑：docker ps | grep pg-test。如果容器没起，docker start pg-test。如果容器起了但端口没映射，检查docker run时的-p 5432:5432。还有一种情况是 Postgres 默认只监听 localhost，容器内没问题，但如果你连的是远程库，需要改postgresql.conf的listen_addresses和pg_hba.conf的访问规则。

报错二：pymysql.err.OperationalError: (1045, "Access denied for user 'root'@'localhost' (using password: YES)")

Mysql 密码错误或用户权限不足。先确认MYSQL_ROOT_PASSWORD环境变量和代码里写的一致。如果容器是之前起的，密码可能没生效，删掉容器重建：docker rm -f mysql-test再重新docker run。另外，Mysql 8 默认认证插件是caching_sha2_password，PyMySQL 新版本支持，如果报认证插件错误，升级 PyMySQL：pip install -U PyMySQL。

报错三：pymongo.errors.ServerSelectionTimeoutError: 127.0.0.1:27017: [Errno 111] Connection refused

Mongo 连接被拒。确认容器在跑，端口映射正确。Mongo 默认不需要认证，如果你开了认证，连接串要写成mongodb://user:pass@host:27017/。另外，Mongo 7 默认绑定127.0.0.1，容器内没问题，远程连接需要改配置。

报错四：requests.exceptions.HTTPError: 401 Client Error: Unauthorized

这是 TaoToken API 调用返回 401，说明 Key 无效或没传。检查三点：环境变量TAOTOKEN_API_KEY是否设置且生效（echo $TAOTOKEN_API_KEY）；请求头Authorization是否是Bearer sk-xxx格式，注意 Bearer 后面有空格；Key 是否复制完整，有没有多余空格。如果都正确还报 401，去控制台确认 Key 是否被禁用或删除，API Keys 页面地址是 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。

报错五：psycopg2.ProgrammingError: cursor already closed或连接池耗尽

这是游标或连接没关导致的。每次操作完，务必cursor.close()和conn.close()。如果用了连接池，用with语句自动管理：

with psycopg2.connect(...) as conn: with conn.cursor() as cur: cur.execute("SELECT 1")

with块结束自动关闭，不用手动调 close。Mongo 的MongoClient是线程安全的，全局一个实例即可，不用每次 close，但进程退出时要client.close()。

报错六：UnicodeDecodeError或中文乱码

Mysql 建表时字符集要指定utf8mb4，连接时charset="utf8mb4"。Postgres 默认 UTF-8，一般没问题。Mongo 存中文没问题，但如果从其他库读出来写进去，注意编码转换。

报错七：local proxy failed或连接超时

如果你在代码里配了代理，但代理不可用，会报这个。检查环境变量HTTP_PROXY、HTTPS_PROXY是否设置，不需要就 unset。TaoToken 的 API 调用如果超时，先确认 Base URL 是https://taotoken.net/api，然后加timeout=30参数，避免无限等待。

排查完这些，基本能覆盖 90% 的常见问题。如果还遇到其他报错，去 TaoToken 的接入文档查一下，地址是 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。

6. 统一 Key 通道实战：数据库操作与 API 调用共用一套配置

最后这一节，把数据库操作和 TaoToken API 调用串起来，演示一个完整的小项目：从 Postgres 读用户数据，调模型生成摘要，写回 Mongo。这样你能看到统一 Key 通道的实际价值。

先准备数据。在 Postgres 里建表插几条：

import psycopg2 conn = psycopg2.connect(host="127.0.0.1", port=5432, database="testdb", user="postgres", password="postgres") cur = conn.cursor() cur.execute("CREATE TABLE IF NOT EXISTS articles (id SERIAL PRIMARY KEY, title TEXT, content TEXT)") cur.execute("INSERT INTO articles (title, content) VALUES (%s, %s)", ("测试文章", "这是一篇关于数据库连接的文章内容。")) conn.commit() cur.close() conn.close()

然后写主流程：

import os import requests import psycopg2 from pymongo import MongoClient TAOTOKEN_KEY = os.getenv("TAOTOKEN_API_KEY") TAOTOKEN_URL = "https://taotoken.net/api/v1/chat/completions" def summarize(text): headers = {"Authorization": f"Bearer {TAOTOKEN_KEY}", "Content-Type": "application/json"} payload = { "model": "gpt-4o-mini", "messages": [{"role": "user", "content": f"用一句话总结：{text}"}] } resp = requests.post(TAOTOKEN_URL, headers=headers, json=payload, timeout=30) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"] def main(): # 从 Postgres 读 conn = psycopg2.connect(host="127.0.0.1", port=5432, database="testdb", user="postgres", password="postgres") cur = conn.cursor() cur.execute("SELECT id, title, content FROM articles LIMIT 1") row = cur.fetchone() cur.close() conn.close() if not row: print("没有数据") return article_id, title, content = row summary = summarize(content) print(f"摘要：{summary}") # 写回 Mongo client = MongoClient("mongodb://127.0.0.1:27017/") col = client["testdb"]["summaries"] col.insert_one({"article_id": article_id, "title": title, "summary": summary}) client.close() print("已写入 Mongo") if __name__ == "__main__": main()

这个流程里，数据库连接和 API 调用各管各的，但 Key 统一从环境变量读。如果 TaoToken Key 轮换，只改一个地方。如果数据库密码变了，也只改配置。这就是统一 Key 通道的好处。

如果你要做更复杂的 Agent 任务，比如让模型根据数据库查询结果决定下一步操作，可以看 TaoToken 的 Coding Plan，地址是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。长期编码类任务用这个比较划算。

最后提醒一句：数据库连接用完一定要关，游标也一样。Mongo 的客户端全局一个实例就行，不用频繁创建销毁。TaoToken 的 API 调用加超时参数，避免网络抖动导致程序卡死。这些细节做好了，三种数据库的日常操作基本不会出问题。