Python技术交互实战：数据库、API与系统运维
Python与其他技术交互实战指南
Python作为一门"胶水语言"，其强大之处在于能轻松与其他技术栈交互。本文将深入探讨Python在数据库、网络API和系统运维三大领域的交互技术。
一、数据库交互
1. ORM框架
SQLAlchemy示例
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
age = Column(Integer)
# 连接数据库
engine = create_engine('sqlite:///example.db')
Base.metadata.create_all(engine)
# 创建会话
Session = sessionmaker(bind=engine)
session = Session()
# 增删改查操作
new_user = User(name='Alice', age=25)
session.add(new_user)
session.commit()
users = session.query(User).filter(User.age > 20).all()实践建议：
- 对于复杂查询，优先使用ORM提供的查询接口而非原生SQL
- 批量操作时考虑使用
bulk_save_objects提高性能 - 生产环境建议配置连接池
Django ORM特点
# models.py
from django.db import models
class Product(models.Model):
name = models.CharField(max_length=100)
price = models.DecimalField(max_digits=10, decimal_places=2)
# 查询示例
cheap_products = Product.objects.filter(price__lt=100).order_by('name')2. 原生数据库驱动
psycopg2 (PostgreSQL)
import psycopg2
conn = psycopg2.connect(
host="localhost",
database="mydb",
user="user",
password="password"
)
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE age > %s", (20,))
rows = cursor.fetchall()
# 使用上下文管理器确保资源释放
with conn.cursor() as cursor:
cursor.execute("INSERT INTO users (name, age) VALUES (%s, %s)",
("Bob", 30))
conn.commit()性能优化技巧：
- 使用预编译语句(prepared statements)防止SQL注入
- 大批量插入时使用
cursor.executemany() - 考虑使用连接池如
psycopg2.pool
3. NoSQL交互
MongoDB (PyMongo)
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['users']
# 插入文档
user = {"name": "Alice", "age": 25, "hobbies": ["coding", "reading"]}
insert_result = collection.insert_one(user)
# 查询
for doc in collection.find({"age": {"$gt": 20}}):
print(doc)Redis (redis-py)
import redis
r = redis.Redis(host='localhost', port=6379, db=0)
# 字符串操作
r.set('foo', 'bar')
value = r.get('foo')
# 哈希表
r.hset('user:1000', mapping={'name': 'Alice', 'age': '25'})
user = r.hgetall('user:1000')二、网络与API交互
1. HTTP请求
requests库高级用法
import requests
# 带认证的请求
response = requests.get(
'https://api.github.com/user',
auth=('username', 'password'),
headers={'Accept': 'application/vnd.github.v3+json'},
params={'since': 1000},
timeout=5
)
# 处理JSON响应
if response.status_code == 200:
data = response.json()
print(data['login'])
# 会话保持
session = requests.Session()
session.get('https://httpbin.org/cookies/set/sessioncookie/123456789')
response = session.get('https://httpbin.org/cookies')最佳实践：
- 总是设置超时(timeout)参数
- 使用会话(Session)对象处理多个请求
- 检查响应状态码(200-400范围内为成功)
aiohttp异步请求
import aiohttp
import asyncio
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
html = await fetch(session, 'http://python.org')
print(html[:100])
loop = asyncio.get_event_loop()
loop.run_until_complete(main())2. RESTful API开发
Flask-RESTful示例
from flask import Flask
from flask_restful import Api, Resource, reqparse
app = Flask(__name__)
api = Api(app)
class UserAPI(Resource):
def __init__(self):
self.parser = reqparse.RequestParser()
self.parser.add_argument('name', type=str, required=True)
self.parser.add_argument('age', type=int)
def get(self, user_id):
return {'user': user_id, 'name': 'Alice'}
def put(self, user_id):
args = self.parser.parse_args()
return {'user': user_id, 'name': args['name']}
api.add_resource(UserAPI, '/users/<int:user_id>')
if __name__ == '__main__':
app.run(debug=True)DRF (Django REST Framework)
# serializers.py
from rest_framework import serializers
from .models import Product
class ProductSerializer(serializers.ModelSerializer):
class Meta:
model = Product
fields = ['id', 'name', 'price']
# views.py
from rest_framework import generics
from .models import Product
from .serializers import ProductSerializer
class ProductList(generics.ListCreateAPIView):
queryset = Product.objects.all()
serializer_class = ProductSerializer
class ProductDetail(generics.RetrieveUpdateDestroyAPIView):
queryset = Product.objects.all()
serializer_class = ProductSerializer3. WebSocket通信
websockets库示例
# 服务端
import asyncio
import websockets
async def echo(websocket, path):
async for message in websocket:
print(f"Received: {message}")
await websocket.send(f"Echo: {message}")
start_server = websockets.serve(echo, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
# 客户端
async def hello():
async with websockets.connect("ws://localhost:8765") as websocket:
await websocket.send("Hello WebSocket!")
response = await websocket.recv()
print(response)
asyncio.get_event_loop().run_until_complete(hello())三、系统与运维自动化
1. 脚本自动化
subprocess最佳实践
import subprocess
# 安全执行命令
result = subprocess.run(
['ls', '-l'],
capture_output=True,
text=True,
check=True # 检查返回码
)
print(result.stdout)
# 管道操作
ps = subprocess.Popen(['ps', '-aux'], stdout=subprocess.PIPE)
grep = subprocess.Popen(['grep', 'python'], stdin=ps.stdout, stdout=subprocess.PIPE)
ps.stdout.close()
output = grep.communicate()[0]安全提示：
- 避免使用
shell=True参数，防止命令注入 - 使用
shlex.quote()处理用户输入的命令参数 - 考虑使用
timeout参数防止进程挂起
2. 日志管理
logging模块配置
import logging
from logging.handlers import RotatingFileHandler
# 创建logger
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)
# 创建处理器
console_handler = logging.StreamHandler()
file_handler = RotatingFileHandler(
'app.log', maxBytes=1024*1024, backupCount=5
)
# 设置格式
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)
# 添加处理器
logger.addHandler(console_handler)
logger.addHandler(file_handler)
# 使用日志
logger.debug('Debug message')
logger.error('Error occurred', exc_info=True)日志策略：
- 开发环境使用DEBUG级别，生产环境使用INFO或WARNING
- 使用RotatingFileHandler或TimedRotatingFileHandler防止日志文件过大
- 敏感信息(如密码)不应记录到日志中
3. 环境配置
python-dotenv示例
# .env文件内容
# DB_HOST=localhost
# DB_PORT=5432
# DB_USER=admin
# DB_PASS=secret
from dotenv import load_dotenv
import os
load_dotenv() # 加载.env文件
db_config = {
'host': os.getenv('DB_HOST'),
'port': os.getenv('DB_PORT'),
'user': os.getenv('DB_USER'),
'password': os.getenv('DB_PASS')
}
print(db_config)安全建议：
- 将.env文件加入.gitignore
- 生产环境使用专门的配置管理系统(如Vault)
- 敏感变量应设置适当权限(如600)
四、交互技术选型指南
技术选型建议：
数据库：
- 事务需求强 → PostgreSQL + SQLAlchemy
- 灵活Schema → MongoDB + PyMongo
- 缓存场景 → Redis + redis-py
网络通信：
- 简单API → requests
- 高并发 → aiohttp
- 实时应用 → websockets
系统运维：
- 简单任务 → subprocess
- 复杂编排 → Ansible API
- 跨平台 → paramiko(SSH)
通过掌握这些交互技术，Python开发者可以轻松构建连接各种组件的强大应用系统。