version: '3.8'

services:
  postgres:
    image: postgres:15
    container_name: my_postgres
    environment:
      POSTGRES_USER: myuser
      POSTGRES_PASSWORD: mypassword
      POSTGRES_DB: mydb
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7
    container_name: my_redis
    ports:
      - "6379:6379"

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.11.0
    container_name: my_es
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false   # 本地测试关闭安全认证
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ports:
      - "9200:9200"
      - "9300:9300"

volumes:
  pgdata:

docker-compose up -d

# 查看容器状态
docker ps

# 检查 PostgreSQL
docker exec -it my_postgres pg_isready -U myuser

# 检查 Elasticsearch（浏览器访问）
# 打开 http://localhost:9200，看到 JSON 响应即正常

# 检查 Redis
docker exec -it my_redis redis-cli ping   # 应返回 PONG

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

uv --version

# 初始化 pyproject.toml
uv init

# 创建虚拟环境（默认在 .venv 目录）
uv venv

# 安装依赖
uv add elasticsearch redis psycopg2-binary

source .venv/Scripts/activate

where python

docker exec -it my_postgres psql -U myuser -d mydb

-- 查看所有表
\dt

-- 查看 products 表结构（字段、类型、约束）
\d products

-- 查看表中的所有数据
SELECT * FROM products;

http://localhost:9200/products/_search?pretty

# 查看所有索引
curl -X GET "http://localhost:9200/_cat/indices?v"

# 查看 products 索引的映射结构（相当于表结构）
curl -X GET "http://localhost:9200/products/_mapping?pretty"

# 查询所有数据（默认10条）
curl -X GET "http://localhost:9200/products/_search?pretty"

# 查询更多数据（size=100）
curl -X GET "http://localhost:9200/products/_search?pretty&size=100"

docker exec -it my_redis redis-cli

# 查看所有键
KEYS *

# 查看 product:p999 的值（JSON 字符串）
GET product:p999

# 查看键的剩余生存时间（TTL，单位秒）
TTL product:p999

# 退出
exit

import redis
r = redis.Redis.from_url("redis://localhost:6379", decode_responses=True)
r.get("product:p999")

database_learn/
├── docker-compose.yml              # Docker 编排文件
├── pyproject.toml                  # 项目依赖清单（uv 管理）
├── uv.lock                         # 锁定文件
├── README.md
├── src/
│   └── database_learn/             # 核心业务包
│       ├── __init__.py
│       ├── config.py               # 连接配置
│       └── service.py              # ProductService 业务类
├── scripts/
│   └── demo.py                     # 完整流程演示脚本
└── docs/
    └── learning-wiki.md            # 本文档

PG_DSN = "dbname=mydb user=myuser password=mypassword host=localhost port=5432"
REDIS_URL = "redis://localhost:6379"
ES_HOST = "http://localhost:9200"

import json
import psycopg2
import redis
from elasticsearch import Elasticsearch, NotFoundError
from .config import PG_DSN, REDIS_URL, ES_HOST

# 全局连接（演示简化，生产应使用连接池）
pg_conn = psycopg2.connect(PG_DSN)
pg_conn.autocommit = True
redis_client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
es_client = Elasticsearch(ES_HOST)


def init_db():
    """初始化 PostgreSQL 表"""
    with pg_conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS products (
                id VARCHAR(50) PRIMARY KEY,
                name TEXT NOT NULL,
                price DECIMAL(10,2),
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
    print("✅ PostgreSQL 表已准备")


def init_es():
    """初始化 Elasticsearch 索引和映射"""
    if not es_client.indices.exists(index="products"):
        es_client.indices.create(index="products", body={
            "mappings": {
                "properties": {
                    "id": {"type": "keyword"},
                    "name": {"type": "text"},
                    "price": {"type": "float"}
                }
            }
        })
        print("✅ Elasticsearch 索引已创建")
    else:
        print("ℹ️  ES 索引已存在")


class ProductService:
    def __init__(self, cache_ttl=30):
        self.cache_ttl = cache_ttl

    # ---------- 写操作 ----------

    def update_product(self, product_id: str, name: str, price: float):
        """写入/更新：DB → ES → 删除缓存（Cache Invalidation）"""
        # 1. 写 PostgreSQL（UPSERT）
        with pg_conn.cursor() as cur:
            cur.execute(
                "INSERT INTO products (id, name, price) VALUES (%s, %s, %s) "
                "ON CONFLICT (id) DO UPDATE SET "
                "  name = EXCLUDED.name, "
                "  price = EXCLUDED.price, "
                "  updated_at = CURRENT_TIMESTAMP",
                (product_id, name, price)
            )
        print(f"📝 DB 已更新: {product_id}")

        # 2. 写 Elasticsearch
        es_client.index(index="products", id=product_id, body={
            "id": product_id,
            "name": name,
            "price": price
        })
        print(f"📄 ES 已索引: {product_id}")

        # 3. 删除 Redis 缓存（使旧数据失效）
        cache_key = f"product:{product_id}"
        redis_client.delete(cache_key)
        print(f"🗑️  缓存已删除: {cache_key}")

    # ---------- 读操作 ----------

    def get_product(self, product_id: str):
        """读取：缓存 → ES → DB（Cache-Aside 三级回源）"""
        cache_key = f"product:{product_id}"

        # --- 第1层：查 Redis 缓存 ---
        cached = redis_client.get(cache_key)
        if cached:
            print(f"✅ 缓存命中: {cache_key}")
            return json.loads(cached)

        # --- 第2层：查 Elasticsearch ---
        try:
            res = es_client.get(index="products", id=product_id)
            data = res["_source"]
            print(f"🔍 ES 查询成功: {product_id}")
        except NotFoundError:
            # --- 第3层：查 PostgreSQL（DB 兜底） ---
            print(f"⚠️  ES 未命中，回查 PostgreSQL: {product_id}")
            with pg_conn.cursor() as cur:
                cur.execute(
                    "SELECT id, name, price FROM products WHERE id = %s",
                    (product_id,)
                )
                row = cur.fetchone()
                if not row:
                    print(f"❌ 商品 {product_id} 不存在（任何层级均无数据）")
                    return None
                data = {"id": row[0], "name": row[1], "price": float(row[2])}

            # 补偿写入 ES（保证 ES 数据最终一致）
            es_client.index(index="products", id=product_id, body=data)
            print(f"📄 从 DB 补偿写入 ES: {product_id}")

        # --- 回写 Redis 缓存（设置 TTL） ---
        redis_client.setex(cache_key, self.cache_ttl, json.dumps(data))
        print(f"💾 缓存已写入: {cache_key} (TTL={self.cache_ttl}s)")
        return data

import json
from database_learn.service import ProductService, init_db, init_es, pg_conn, redis_client, es_client


def main():
    print("🚀 初始化数据库和索引...")
    init_db()
    init_es()

    service = ProductService()
    test_id = "p999"

    print("\n--- 1. 写入商品 ---")
    service.update_product(test_id, "Python 从入门到实践", 59.90)

    print("\n--- 2. 第一次读取（缓存未命中，走 ES） ---")
    data = service.get_product(test_id)
    print(f"   结果: {data}")

    print("\n--- 3. 第二次读取（应该命中缓存） ---")
    data = service.get_product(test_id)
    print(f"   结果: {data}")

    print("\n--- 4. 更新商品价格（触发缓存删除） ---")
    service.update_product(test_id, "Python 从入门到实践", 79.90)

    print("\n--- 5. 更新后再次读取（缓存失效，走 ES 拿新数据） ---")
    data = service.get_product(test_id)
    print(f"   结果: {data}")

    # 验证数据库
    print("\n--- 6. 验证数据库中的最终数据 ---")
    with pg_conn.cursor() as cur:
        cur.execute("SELECT id, name, price FROM products WHERE id = %s", (test_id,))
        row = cur.fetchone()
        print(f"   PostgreSQL: id={row[0]}, name={row[1]}, price={row[2]}")

    print("\n--- 7. 验证 Redis 缓存状态 ---")
    cached = redis_client.get(f"product:{test_id}")
    if cached:
        print(f"   Redis 缓存存在: {json.loads(cached)}")
    else:
        print("   Redis 缓存已过期或不存在")

    print("\n--- 8. 验证 Elasticsearch 数据 ---")
    res = es_client.get(index="products", id=test_id)
    print(f"   Elasticsearch: {res['_source']}")

    print("\n🎉 全部验证通过！")


if __name__ == "__main__":
    main()

# 确保 Docker 容器正在运行
docker ps

# 激活虚拟环境（如果尚未激活）
source .venv/Scripts/activate

# 运行演示脚本
python scripts/demo.py

🚀 初始化数据库和索引...
✅ PostgreSQL 表已准备
ℹ️  ES 索引已存在

--- 1. 写入商品 ---
📝 DB 已更新: p999
📄 ES 已索引: p999
🗑️  缓存已删除: product:p999

--- 2. 第一次读取（缓存未命中，走 ES） ---
🔍 ES 查询成功: p999
💾 缓存已写入: product:p999 (TTL=30s)
   结果: {'id': 'p999', 'name': 'Python 从入门到实践', 'price': 59.9}

--- 3. 第二次读取（应该命中缓存） ---
✅ 缓存命中: product:p999
   结果: {'id': 'p999', 'name': 'Python 从入门到实践', 'price': 59.9}

--- 4. 更新商品价格（触发缓存删除） ---
📝 DB 已更新: p999
📄 ES 已索引: p999
🗑️  缓存已删除: product:p999

--- 5. 更新后再次读取（缓存失效，走 ES 拿新数据） ---
🔍 ES 查询成功: p999
💾 缓存已写入: product:p999 (TTL=30s)
   结果: {'id': 'p999', 'name': 'Python 从入门到实践', 'price': 79.9}

--- 6. 验证数据库中的最终数据 ---
   PostgreSQL: id=p999, name=Python 从入门到实践, price=79.90

--- 7. 验证 Redis 缓存状态 ---
   Redis 缓存存在: {'id': 'p999', 'name': 'Python 从入门到实践', 'price': 79.9}

--- 8. 验证 Elasticsearch 数据 ---
   Elasticsearch: {'id': 'p999', 'name': 'Python 从入门到实践', 'price': 79.9}

🎉 全部验证通过！