Python 非关系型数据库操作完全指南

pip install redis

import redis

# redis-py 提供两个客户端类：
#   StrictRedis — 命令与官方 Redis 命令一一对应
#   Redis       — StrictRedis 的子类，向后兼容旧版

r = redis.StrictRedis(host='127.0.0.1', port=6379, db=0)
# 默认返回 bytes；设置 decode_responses=True 返回 str
r_str = redis.StrictRedis(host='127.0.0.1', port=6379, db=0, decode_responses=True)

print(r.keys())     # [b'key1', b'key2']
print(r_str.keys()) # ['key1', 'key2']

import redis

pool = redis.ConnectionPool(
    host='127.0.0.1',
    port=6379,
    db=0,
    decode_responses=True,
    max_connections=50
)
r = redis.Redis(connection_pool=pool)  # 多个 Redis 实例可共享同一个池

print(r.keys())

import redis
import time

r = redis.Redis(connection_pool=pool)

# 带过期时间的设置（缓存常用）
r.set('temp_token', 'abc123', ex=10)
print(r.get('temp_token'))  # 'abc123'
time.sleep(10)
print(r.get('temp_token'))  # None — 已过期删除

# 分布式锁模式 — 仅当不存在时设置
acquired = r.set('lock:order:123', 'owner-1', nx=True, ex=30)
if acquired:
    try:
        process_order(123)
    finally:
        r.delete('lock:order:123')

# 计数器
r.incr('page:view:2024-01-01', amount=1)
r.incrbyfloat('score:player1', 3.5)

# 批量操作 — 一次网络往返完成多个命令
r.mset({'user:1:name': 'Alice', 'user:1:email': 'alice@example.com'})
values = r.mget('user:1:name', 'user:1:email')
print(values)  # ['Alice', 'alice@example.com']

# 消息队列 — 生产者-消费者模式
# 生产者
r.rpush('task_queue', 'task_1', 'task_2', 'task_3')

# 消费者 — 阻塞等待，减少空轮询的开销
while True:
    result = r.blpop('task_queue', timeout=5)
    if result:
        key, task = result
        print(f'处理任务: {task}')
    else:
        print('队列为空，等待中...')
        break

# 共同好友
r.sadd('friends:Alice', 'Bob', 'Charlie', 'David')
r.sadd('friends:Bob', 'Alice', 'Charlie', 'Eve')

common = r.sinter('friends:Alice', 'friends:Bob')
print(f'共同好友: {common}')  # {'Charlie'}

# 推荐好友（Bob 的好友中 Alice 没加的）
suggest = r.sdiff('friends:Bob', 'friends:Alice')
print(f'推荐: {suggest}')  # {'Alice', 'Eve'}

# 游戏排行榜
r.zadd('leaderboard', {'Alice': 9500, 'Bob': 8700, 'Charlie': 9200, 'David': 7800})

# Top 3
top3 = r.zrevrange('leaderboard', 0, 2, withscores=True)
print(f'Top 3: {top3}')  # [('Alice', 9500.0), ('Charlie', 9200.0), ('Bob', 8700.0)]

# 查排名
rank = r.zrevrank('leaderboard', 'Bob')
print(f'Bob 排名: {rank + 1}')  # 3

# 加分
r.zincrby('leaderboard', 500, 'David')

# 用户信息存储
r.hset('user:1001', mapping={
    'name': '张三',
    'email': 'zhangsan@example.com',
    'login_count': '42'
})

user = r.hgetall('user:1001')
print(user)  # {'name': '张三', 'email': 'zhangsan@..., 'login_count': '42'}

# 原子递增登录次数
r.hincrby('user:1001', 'login_count', 1)

# 分批扫码大数据量 Hash
for field_batch in r.hscan_iter('large_hash', count=100):
    key, value = field_batch
    process(key, value)

pipe = r.pipeline()

# 链式调用
pipe.set('name', 'jack').set('role', 'admin').incr('counter')
results = pipe.execute()
print(results)  # [True, True, 1]

print(r.get('name'))   # 'jack'
print(r.get('counter'))  # '1'

pip install pymongo

from pymongo import MongoClient

# 连接 MongoDB
client = MongoClient('mongodb://127.0.0.1:27017')

# 获取数据库（不存在则自动创建）
db = client.school

# 获取集合
students = db.students

# 插入单个
students.insert_one({
    'stuid': '2024001',
    'name': '张三',
    'age': 20,
    'scores': {'math': 88, 'english': 92}
})

# 批量插入
students.insert_many([
    {'stuid': '2024002', 'name': '李四', 'age': 22, 'scores': {'math': 76, 'english': 85}},
    {'stuid': '2024003', 'name': '王五', 'age': 21, 'scores': {'math': 95, 'english': 90}},
])

# 查询所有
for student in students.find():
    print(student['name'], student['age'])

# 条件查询
for s in students.find({'age': {'$gte': 21}}):
    print(s['name'])  # 年龄 >= 21

# 投影（只返回指定字段）
for s in students.find({}, {'name': 1, 'age': 1, '_id': 0}):
    print(s)

# 排序与分页
results = (
    students.find()
    .sort('age', -1)    # 按年龄降序
    .limit(10)
    .skip(20)
)

# 更新单条
students.update_one(
    {'stuid': '2024001'},
    {'$set': {'age': 21}}
)

# 更新多条
students.update_many(
    {'age': {'$lt': 22}},
    {'$inc': {'age': 1}}  # 年龄 +1
)

# 删除
students.delete_one({'stuid': '2024003'})
students.delete_many({'age': {'$gt': 25}})

# 统计
count = students.count_documents({'age': {'$gte': 20}})
print(f'符合条件的文档数: {count}')