第3章：数据库架构设计¶

from pymongo import MongoClient

client = MongoClient('mongodb://localhost:27017/')
db = client['mydb']
collection = db['users']

# 插入文档
user = {
    'name': 'John',
    'email': 'john@example.com',
    'age': 30,
    'address': {
        'city': 'Beijing',
        'country': 'China'
    }
}
collection.insert_one(user)

# 查询文档
result = collection.find_one({'name': 'John'})

import redis

r = redis.Redis(host='localhost', port=6379, db=0)

# 字符串
r.set('key', 'value')
value = r.get('key')

# 哈希
r.hset('user:1', 'name', 'John')
r.hset('user:1', 'email', 'john@example.com')
name = r.hget('user:1', 'name')

# 列表
r.lpush('tasks', 'task1')
r.lpush('tasks', 'task2')
task = r.rpop('tasks')

# 集合
r.sadd('tags', 'python')
r.sadd('tags', 'redis')
tags = r.smembers('tags')

# 有序集合
r.zadd('leaderboard', {'user1': 100, 'user2': 200})
top_users = r.zrange('leaderboard', 0, 1, desc=True)

import uuid
from cassandra.cluster import Cluster

cluster = Cluster(['127.0.0.1'])
session = cluster.connect('mykeyspace')

# 创建表
session.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id UUID PRIMARY KEY,
        name TEXT,
        email TEXT
    )
""")

# 插入数据
user_id = uuid.uuid4()
session.execute(
    "INSERT INTO users (id, name, email) VALUES (%s, %s, %s)",
    (user_id, 'John', 'john@example.com')
)

# 按主键查询（Cassandra 要求按分区键查询）
rows = session.execute("SELECT * FROM users WHERE id = %s", (user_id,))
for row in rows:
    print(row.name, row.email)

# 如需按非主键字段查询，需创建二级索引
# CREATE INDEX IF NOT EXISTS idx_users_name ON users(name);
# rows = session.execute("SELECT * FROM users WHERE name = 'John'")

from neo4j import GraphDatabase

driver = GraphDatabase.driver("bolt://localhost:7687",
                              auth=("neo4j", "password"))

def create_person(driver, name):
    with driver.session() as session:
        session.run("CREATE (p:Person {name: $name})", name=name)

def create_friendship(driver, person1, person2):
    with driver.session() as session:
        session.run("""
            MATCH (p1:Person {name: $person1})
            MATCH (p2:Person {name: $person2})
            CREATE (p1)-[:FRIEND]->(p2)
        """, person1=person1, person2=person2)

# 创建节点
create_person(driver, "Alice")
create_person(driver, "Bob")

# 创建关系
create_friendship(driver, "Alice", "Bob")

import random

class DatabaseRouter:
    def __init__(self, master, slaves):
        self.master = master
        self.slaves = slaves

    def get_connection(self, operation):
        if operation == 'read':
            # 随机选择一个从库
            return random.choice(self.slaves)
        else:
            # 写操作使用主库
            return self.master

    def execute(self, operation, sql, params=None):
        conn = self.get_connection(operation)
        cursor = conn.cursor()
        cursor.execute(sql, params or ())
        if operation == 'read':
            return cursor.fetchall()
        else:
            conn.commit()
            return cursor.rowcount

# 使用示例
master = MySQLConnection('master_host')
slaves = [
    MySQLConnection('slave1_host'),
    MySQLConnection('slave2_host'),
]

router = DatabaseRouter(master, slaves)

# 读操作
results = router.execute('read', 'SELECT * FROM users')

# 写操作
router.execute('write', 'INSERT INTO users (name) VALUES (%s)', ('John',))

# Hash分片示例
def get_shard(user_id, num_shards):
    return user_id % num_shards

def get_table_name(user_id, num_shards):
    shard_id = get_shard(user_id, num_shards)
    return f"users_{shard_id}"

# 使用示例
user_id = 12345
num_shards = 4
table_name = get_table_name(user_id, num_shards)
# table_name = "users_1"

# 查询（表名来自内部逻辑计算，非用户输入，因此可安全拼接）
# ⚠️ 注意：若table_name可能来自用户输入，必须使用白名单验证
sql = f"SELECT * FROM {table_name} WHERE id = %s"