每日python库：web开发（6）
|
Chacantingtui
|
1,107
|
python

975 字
|
6 分钟

Python库的多样性使得它们可以在不同的应用场景中发挥作用，不同场景下同一个库可能拥有不同的功能，本系列未来会有不同的分类，可能会出现相同的库。数量很多，慢慢整理。文中的示例都是最基础的用法，有些就顺手写了，有些就丢给Ai，看个乐就行。

1. python-multipart – 文件和表单处理

功能: 用于处理 HTTP multipart/form-data 请求，支持文件上传和表单数据解析。

使用场景:

  • 处理文件上传的 Web 应用。
  • 与 Flask、Django 等框架集成处理表单数据。
  • 需要解析多部分请求的任何 HTTP 服务。

示例:

from werkzeug.datastructures import FileStorage
from werkzeug.formparser import parse_form_data

# 示例请求对象
class Request:
def __init__(self, stream, content_type):
self.stream = stream
self.content_type = content_type

# 创建一个模拟的请求对象
request = Request(b'--boundary\r\nContent-Disposition: form-data; name="file"; filename="test.txt"\r\n\r\nHello, World!\r\n--boundary--', 'multipart/form-data; boundary=boundary')

data, files = parse_form_data(request, stream_factory=FileStorage)

# 处理文件
for key, file in files.items():
print(f"File name: {file.filename}")
print(f"File content: {file.read().decode('utf-8')}")

2. aiofiles – 异步文件操作支持

功能: 为异步编程提供文件 I/O 支持，用于与 asyncio 兼容的文件操作。

使用场景:

  • 需要异步文件读写的高并发应用。
  • 与异步框架如 FastAPI、AIOHTTP 集成。
  • 提高 I/O 密集型任务的性能。

示例:

import asyncio
import aiofiles

async def main():
async with aiofiles.open('example.txt', mode='w') as f:
await f.write('Hello, World!')

async with aiofiles.open('example.txt', mode='r') as f:
content = await f.read()
print(content)

asyncio.run(main())

3. gridfs – MongoDB 的文件存储系统

功能: GridFS 是 MongoDB 的一部分，用于存储和检索大文件。

使用场景:

  • 需要在 MongoDB 中存储大文件（>16MB）的应用。
  • 构建内容管理系统或文件服务。
  • 需要高效、可扩展文件存储的项目。

示例:

from pymongo import MongoClient
from gridfs import GridFS

# 连接到 MongoDB
client = MongoClient('localhost', 27017)
db = client.test_database
fs = GridFS(db)

# 写入文件
with open('example.txt', 'rb') as f:
content = f.read()
fs.put(content, filename="example.txt")

# 读取文件
file = fs.find_one({"filename": "example.txt"})
with open('downloaded_example.txt', 'wb') as f:
f.write(file.read())

4. Celery – 分布式任务队列

功能: Celery 是一个异步任务队列/作业队列，基于分布式消息传递。

使用场景:

  • 处理后台任务、定时任务。
  • 分布式任务处理和负载均衡。
  • 与 Django、Flask 等 Web 框架集成。

示例:

from celery import Celery

app = Celery('tasks', broker='pyamqp://guest@localhost//')

@app.task
def add(x, y):
return x + y

# 运行 Celery 工作进程
# celery -A tasks worker --loglevel=info

5. Dramatiq – 高性能任务队列

功能: Dramatiq 是一个快速、可靠的任务处理库，支持分布式处理。

使用场景:

  • 需要高性能任务队列的应用。
  • 微服务架构中的任务处理。
  • 需要任务优先级和重试机制的项目。

示例:

from dramatiq import GenericActor

class ExampleActor(GenericActor):
def perform(self, x, y):
print(f"Adding {x} + {y}")
return x + y

# 在主程序中
from dramatiq.brokers.rabbitmq import RabbitmqBroker
from dramatiq.results import Results
from dramatiq.results.backends import RedisBackend

broker = RabbitmqBroker()
broker.add_middleware(Results(backend=RedisBackend()))

# 发送任务到队列
ExampleActor.send(1, 2)

6. RQ (Redis Queue) – 基于 Redis 的简单任务队列

功能: RQ 是一个简单、轻量级的任务队列，依赖于 Redis。

使用场景:

  • 需要快速设置任务队列的项目。
  • 简单任务处理和调度。
  • 与 Python Web 框架集成。

示例:

from redis import Redis
from rq import Queue

# 连接到 Redis
redis_conn = Redis()
q = Queue(connection=redis_conn)

# 定义任务
def example_task(x, y):
return x + y

# 将任务加入队列
job = q.enqueue(example_task, 2, 3)

7. Huey – 小型任务队列框架

功能: Huey 是一个轻量级的任务队列，支持多种后端存储。

使用场景:

  • 需要一个简单、易用的任务队列。
  • 适用于小型项目或原型开发。
  • 与 Django 集成。

示例:

from huey import RedisHuey

huey = RedisHuey('my_app')

@huey.task()
def add(x, y):
return x + y

# 使用任务
result = add(1, 2)
print(result.get()) # 等待任务完成并获取结果

8. APScheduler – 调度和执行定时任务

功能: APScheduler 是一个 Python 库，用于调度任务，支持多种调度方式。

使用场景:

  • 定时执行任务。
  • 需要复杂调度逻辑的应用。
  • 与 Web 框架集成以实现定时作业。

示例:

from apscheduler.schedulers.background import BackgroundScheduler

def job():
print("Scheduled job executed!")

scheduler = BackgroundScheduler()
scheduler.add_job(job, 'interval', seconds=5)
scheduler.start()

# 保持程序运行
try:
while True:
pass
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()

9. Werkzeug – WSGI 工具和调试器

功能: Werkzeug 提供了 WSGI 工具集和一个强大的调试器，常用于 Web 框架的底层。

使用场景:

  • 构建自定义 WSGI 中间件。
  • 开发需要 WSGI 支持的 Web 应用。
  • 与 Flask 等框架集成。

示例:

from werkzeug.wrappers import Request, Response

@Request.application
def application(request):
return Response('Hello, World!')

if __name__ == '__main__':
from werkzeug.serving import run_simple
run_simple('localhost', 5000, application)

10. pytest – 强大的测试框架

功能: pytest 是一个功能强大的测试框架，支持参数化、插件和复杂的测试结构。

使用场景:

  • 编写和运行单元测试、集成测试。
  • 需要灵活、可扩展的测试框架。
  • 与其他测试工具和框架集成。

示例:

# test_example.py
def inc(x):
return x + 1

def test_inc():
assert inc(3) == 4

# 运行测试
# pytest test_example.py
每日python库
上一篇
下一篇