Memos SQLite → PostgreSQL 迁移指南

稚

2026-08-02

教程

版本：v0.3 | 环境：pgsql v18.0（均为 Docker 部署）

前提条件

备份原有的 SQLite 数据库文件
拷贝备份到本地
新建 PostgreSQL 数据库
启动 Memos 项目并连接 PostgreSQL 完成初始化

Docker Compose 配置

services:
  memos:
    image: neosmemo/memos:0.30
    container_name: memos
    # 1000 为我的用户，988 为 docker 组
    user: "1000:988"
    volumes:
      - ./:/var/opt/memos
    ports:
      - 5230:5230
    restart: unless-stopped
    environment:
      # PostgreSQL 驱动
      MEMOS_DRIVER: postgres
      # PostgreSQL 连接信息
      MEMOS_DSN: "postgres://用户:密码@数据库ip:端口/库名?sslmode=disable"

验证连接成功

运行 docker compose logs，看到以下输出表示连接成功：

memos  | Memos 0.30.0 started successfully!
memos  | Data directory: /var/opt/memos
memos  | Database driver: postgres
memos  | Server running on port 1234
memos  | Access your memos at: http://localhost:1234
memos  | Access mode: private
memos  |
memos  | Documentation: https://usememos.com
memos  | Source code: https://github.com/usememos/memos
memos  |
memos  | Happy note-taking!

数据迁移脚本

以下脚本由 AI 生成，已实现：

memo / attachment 数据全部迁移

attachment.blob 留在 SQLite 不迁移（文件本身上传 S3）

新库 attachment 表：storage_type='S3' + presigned URL + payload s3Object JSON

迁移脚本 (`migrate.py`)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
memos: SQLite -> PostgreSQL 迁移
- memo / attachment 数据全部迁过来
- attachment.blob 留在 SQLite 不迁
- 文件本身传 S3
- 新库 attachment: storage_type='S3' + presigned URL + payload s3Object JSON
"""

import sqlite3
import json
import logging
import os
import sys
import time
from datetime import datetime, timezone
from io import BytesIO

import boto3
import psycopg2
from botocore.config import Config as BotoConfig
from botocore.exceptions import ClientError

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    stream=sys.stderr,
)
LOG = logging.getLogger("memos-migrate")


# ============================================================
# 配置
# ============================================================
CONFIG = {
    # ---- Source: SQLite ----
    "sqlite_path": "",

    # ---- Target: PostgreSQL ----
    "pg_host":     "",
    "pg_port":     5432,
    "pg_user":     "",
    "pg_password": "",
    "pg_dbname":   "",

    # ---- S3 ----
    "s3_endpoint":    "",
    "s3_region":      "",
    "s3_bucket":      "",
    "s3_access_key":  "",
    "s3_secret_key":  "",
    "s3_key_prefix":  "",
    "s3_expires":     432000,

    # ---- 用户处理模式 ----
    # "auto"  : 按 username 创 user，复用旧密码哈希（user_id 保持一致）
    # "skip"  : 不迁 user，新库需要自己创
    # "manual": 用下面的 user_map 手动映射
    # "match" : 按 username 匹配，PG 有同名则复用，没有则 INSERT
    "user_mode": "match",   # auto/skip/manual/match
    "user_map":  {},
}
# ============================================================


def E(msg):
    print(msg, file=sys.stderr, flush=True)


def make_s3(cfg):
    return boto3.client(
        "s3",
        endpoint_url=cfg["s3_endpoint"],
        region_name=cfg["s3_region"],
        aws_access_key_id=cfg["s3_access_key"],
        aws_secret_access_key=cfg["s3_secret_key"],
        config=BotoConfig(signature_version="s3v4"),
    )


def connect_sqlite(path):
    if not os.path.exists(path):
        raise SystemExit(f"X SQLite file not found: {path}")
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row
    return conn


def connect_pg(cfg):
    return psycopg2.connect(
        host=cfg["pg_host"], port=cfg["pg_port"],
        user=cfg["pg_user"], password=cfg["pg_password"],
        dbname=cfg["pg_dbname"],
    )


def _now_iso_nano_z():
    """返回 memos 期望的纳秒精度 + Z 时区格式，如 2026-08-02T06:08:04.729207108Z"""
    import time as _t
    ns = _t.time_ns()
    sec = ns // 1_000_000_000
    nano = ns % 1_000_000_000
    dt = _t.gmtime(sec)
    return f"{_t.strftime('%Y-%m-%dT%H:%M:%S', dt)}.{nano:09d}Z"


def build_payload(cfg, s3_key):
    """memos v0.30.x 期望的 payload 结构"""
    endpoint = cfg["s3_endpoint"].rstrip("/") + "/"
    return json.dumps({
        "s3Object": {
            "s3Config": {
                "accessKeyId":            cfg["s3_access_key"],
                "accessKeySecret":        cfg["s3_secret_key"],
                "endpoint":               endpoint,
                "region":                 cfg["s3_region"],
                "bucket":                 cfg["s3_bucket"],
                "usePathStyle":           True,
                "insecureSkipTlsVerify":  True,
            },
            "key": s3_key,
            "lastPresignedTime": _now_iso_nano_z(),
        }
    }, ensure_ascii=False)


def migrate_users(sqlite_conn, pg_conn, cfg):
    src = sqlite_conn.cursor()
    src.execute("""
        SELECT id, username, role, email, nickname, password_hash,
               avatar_url, description, row_status, created_ts, updated_ts
        FROM user
    """)
    rows = src.fetchall()
    LOG.info("source user count: %d", len(rows))

    if cfg["user_mode"] == "skip":
        LOG.warning("user_mode=skip, skip user migration. "
                    "you need to create users in new PG and provide user_map.")
        return {}

    # 读 PG 里已存在的 user（按 username 建索引）
    pg = pg_conn.cursor()
    pg.execute('SELECT id, username FROM "user"')
    pg_user_by_name = {row[1]: row[0] for row in pg.fetchall()}
    LOG.info("PG existing user count: %d", len(pg_user_by_name))

    old_to_new = {}
    new_inserted = 0
    matched = 0
    for r in rows:
        sqlite_uid = r["id"]
        sqlite_username = r["username"]

        if cfg["user_mode"] == "manual":
            new_id = cfg["user_map"].get(sqlite_uid)
            if not new_id:
                LOG.warning("user %s (id=%d) no mapping, skip",
                            sqlite_username, sqlite_uid)
                continue
            old_to_new[sqlite_uid] = new_id
        elif cfg["user_mode"] == "match":
            # 按 username 匹配：PG 里有同名 → 用 PG 的 id；没有 → INSERT
            if sqlite_username in pg_user_by_name:
                old_to_new[sqlite_uid] = pg_user_by_name[sqlite_username]
                matched += 1
                LOG.info("  [MATCH] sqlite user '%s' (id=%d) -> pg user id=%d",
                         sqlite_username, sqlite_uid, pg_user_by_name[sqlite_username])
            else:
                # PG 里没这个 username，从 SQLite 拷贝过来（用 SQLite 原 id）
                pg.execute("""
                    INSERT INTO "user" (id, username, role, email, nickname,
                                        password_hash, avatar_url, description,
                                        row_status, created_ts, updated_ts)
                    VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                    ON CONFLICT (id) DO NOTHING
                """, (sqlite_uid, sqlite_username, r["role"], r["email"], r["nickname"],
                      r["password_hash"], r["avatar_url"], r["description"],
                      r["row_status"], r["created_ts"], r["updated_ts"]))
                old_to_new[sqlite_uid] = sqlite_uid
                new_inserted += 1
                LOG.info("  [NEW]   sqlite user '%s' (id=%d) -> inserted to PG",
                         sqlite_username, sqlite_uid)
        else:  # auto
            pg.execute("""
                INSERT INTO "user" (id, username, role, email, nickname,
                                    password_hash, avatar_url, description,
                                    row_status, created_ts, updated_ts)
                VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
                ON CONFLICT (id) DO NOTHING
            """, (sqlite_uid, sqlite_username, r["role"], r["email"], r["nickname"],
                  r["password_hash"], r["avatar_url"], r["description"],
                  r["row_status"], r["created_ts"], r["updated_ts"]))
            old_to_new[sqlite_uid] = sqlite_uid

    pg_conn.commit()
    if cfg["user_mode"] == "match":
        LOG.info("user map: %d entries (matched=%d, new=%d)",
                 len(old_to_new), matched, new_inserted)
    else:
        LOG.info("user map: %d entries", len(old_to_new))
    return old_to_new


def migrate_memos(sqlite_conn, pg_conn, user_map):
    src = sqlite_conn.cursor()
    src.execute("""
        SELECT id, uid, creator_id, created_ts, updated_ts, row_status,
               content, visibility, pinned, payload
        FROM memo
    """)
    rows = src.fetchall()
    LOG.info("source memo count: %d", len(rows))

    pg = pg_conn.cursor()
    for r in rows:
        new_creator = user_map.get(r["creator_id"], 1)
        pg.execute("""
            INSERT INTO memo (id, uid, creator_id, created_ts, updated_ts,
                              row_status, content, visibility, pinned, payload)
            VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            ON CONFLICT (id) DO UPDATE SET
                content    = EXCLUDED.content,
                creator_id = EXCLUDED.creator_id,
                updated_ts = EXCLUDED.updated_ts
        """, (r["id"], r["uid"], new_creator, r["created_ts"], r["updated_ts"],
              r["row_status"], r["content"], r["visibility"], bool(r["pinned"]),
              r["payload"]))
    pg_conn.commit()
    LOG.info("memo migrated: %d rows", len(rows))


def migrate_attachments(sqlite_conn, pg_conn, s3, user_map, cfg):
    src = sqlite_conn.cursor()
    src.execute("""
        SELECT id, uid, creator_id, created_ts, updated_ts, filename,
               type, size, memo_id, storage_type, reference, payload, blob
        FROM attachment
    """)
    rows = src.fetchall()
    LOG.info("source attachment count: %d", len(rows))

    pg = pg_conn.cursor()
    ok = fail = 0
    for r in rows:
        blob = r["blob"]
        if blob:
            try:
                s3_key = f"{cfg['s3_key_prefix']}{r['created_ts'] // 1000}_{r['filename']}"
                s3.upload_fileobj(
                    BytesIO(bytes(blob)),
                    cfg["s3_bucket"], s3_key,
                    ExtraArgs={"ContentType": r["type"] or "application/octet-stream"},
                )
                presigned_url = s3.generate_presigned_url(
                    "get_object",
                    Params={"Bucket": cfg["s3_bucket"], "Key": s3_key},
                    ExpiresIn=cfg["s3_expires"],
                )
                payload_json = build_payload(cfg, s3_key)
                storage_type = "S3"
                reference = presigned_url
            except ClientError as e:
                LOG.error("[S3 FAIL] attachment id=%s: %s", r["id"], e)
                fail += 1
                continue
        else:
            storage_type = r["storage_type"] or "S3"
            reference = r["reference"] or ""
            payload_json = r["payload"] or "{}"

        new_creator = user_map.get(r["creator_id"], 1)

        pg.execute("""
            INSERT INTO attachment (id, uid, creator_id, created_ts, updated_ts,
                                    filename, type, size, memo_id,
                                    storage_type, reference, payload, blob)
            VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s, NULL)
            ON CONFLICT (id) DO UPDATE SET
                reference    = EXCLUDED.reference,
                payload      = EXCLUDED.payload,
                storage_type = EXCLUDED.storage_type,
                blob         = NULL,
                creator_id   = EXCLUDED.creator_id,
                memo_id      = EXCLUDED.memo_id
        """, (r["id"], r["uid"], new_creator, r["created_ts"], r["updated_ts"],
              r["filename"], r["type"], r["size"], r["memo_id"],
              storage_type, reference, payload_json))
        ok += 1
        if ok % 20 == 0:
            LOG.info("processed %d / %d", ok, len(rows))

    pg_conn.commit()
    LOG.info("attachment migrated: ok=%d fail=%d", ok, fail)


def reset_sequences(pg_conn):
    pg = pg_conn.cursor()
    for seq, table in [
        ("memo_id_seq", "memo"),
        ("attachment_id_seq", "attachment"),
        ("user_id_seq", '"user"'),
    ]:
        try:
            pg.execute(f"SELECT setval('{seq}', COALESCE((SELECT MAX(id) FROM {table}), 1))")
            LOG.info("reset sequence: %s", seq)
        except psycopg2.errors.UndefinedTable:
            LOG.warning("sequence/table %s not found, skip", seq)
    pg_conn.commit()


def main():
    E("=" * 60)
    E("memos: SQLite -> PostgreSQL migration")
    E("  sqlite: " + CONFIG["sqlite_path"])
    E("  pg:     " + CONFIG["pg_host"] + ":" + str(CONFIG["pg_port"]) + "/" + CONFIG["pg_dbname"])
    E("  s3:     " + CONFIG["s3_endpoint"] + "/" + CONFIG["s3_bucket"])
    E("=" * 60)

    sq = connect_sqlite(CONFIG["sqlite_path"])
    pg = connect_pg(CONFIG)
    s3 = make_s3(CONFIG)

    user_map = migrate_users(sq, pg, CONFIG)
    migrate_memos(sq, pg, user_map)
    migrate_attachments(sq, pg, s3, user_map, CONFIG)
    reset_sequences(pg)

    E("=" * 60)
    E("V migration done")
    E("  user mapping: " + str(len(user_map)))
    E("  next steps: archive SQLite, VACUUM FULL on PG, verify counts")
    E("=" * 60)


if __name__ == "__main__":
    main()

S3 凭证轮换脚本

当 S3 存储密钥更换后，数据库中旧的密钥不会自动更新，需要使用此脚本手动更换：

此脚本功能：

UPDATE 所有 attachment.payload 里的 s3Config（accessKeyId/Secret）

同时重新生成 presigned URL（让 reference 用新凭证签名）

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
轮换 S3 凭证：
1. UPDATE 所有 attachment.payload 里的 s3Config（accessKeyId/Secret）
2. 同时重新生成 presigned URL（让 reference 用新凭证签名）
"""
import json
import sys
import time

import boto3
import psycopg2
from botocore.config import Config as BotoConfig
from botocore.exceptions import ClientError

CONFIG = {
    "pg_host":     "",
    "pg_port":     5432,
    "pg_user":     "",
    "pg_password": "",
    "pg_dbname":   "",

    # ---- 新 S3 凭证 ----
    "s3_endpoint":     "",
    "s3_region":       "",
    "s3_bucket":       "",
    "new_access_key":  "",
    "new_secret_key":  "",
    "s3_expires":      432000,
}


def _now_iso_nano_z():
    ns = time.time_ns()
    sec = ns // 1_000_000_000
    nano = ns % 1_000_000_000
    dt = time.gmtime(sec)
    return f"{time.strftime('%Y-%m-%dT%H:%M:%S', dt)}.{nano:09d}Z"


def main():
    print("connect s3 with NEW credentials ...", flush=True)
    s3 = boto3.client(
        "s3",
        endpoint_url=CONFIG["s3_endpoint"],
        region_name=CONFIG["s3_region"],
        aws_access_key_id=CONFIG["new_access_key"],
        aws_secret_access_key=CONFIG["new_secret_key"],
        config=BotoConfig(signature_version="s3v4"),
    )
    # 验证新凭证能访问
    try:
        s3.head_bucket(Bucket=CONFIG["s3_bucket"])
        print(f"V new credentials OK, bucket '{CONFIG['s3_bucket']}' accessible")
    except ClientError as e:
        print(f"X new credentials FAIL: {e}")
        return 1

    conn = psycopg2.connect(
        host=CONFIG["pg_host"], port=CONFIG["pg_port"],
        user=CONFIG["pg_user"], password=CONFIG["pg_password"],
        dbname=CONFIG["pg_dbname"],
    )
    cur = conn.cursor()
    cur.execute("""
        SELECT id, payload, reference
        FROM attachment
        WHERE storage_type = 'S3'
    """)
    rows = cur.fetchall()
    print(f"待更新 {len(rows)} 行")

    if input("输入 YES 继续：").strip() != "YES":
        print("已取消")
        return 1

    endpoint = CONFIG["s3_endpoint"].rstrip("/") + "/"
    ok = fail = 0
    for old_id, payload_str, old_ref in rows:
        try:
            payload = json.loads(payload_str)
            s3_key = payload["s3Object"]["key"]

            # 1. 更新 s3Config
            payload["s3Object"]["s3Config"] = {
                "accessKeyId":           CONFIG["new_access_key"],
                "accessKeySecret":       CONFIG["new_secret_key"],
                "endpoint":              endpoint,
                "region":                CONFIG["s3_region"],
                "bucket":                CONFIG["s3_bucket"],
                "usePathStyle":          True,
                "insecureSkipTlsVerify": True,
            }
            payload["s3Object"]["lastPresignedTime"] = _now_iso_nano_z()

            # 2. 用新凭证重新生成 presigned URL
            new_presigned = s3.generate_presigned_url(
                "get_object",
                Params={"Bucket": CONFIG["s3_bucket"], "Key": s3_key},
                ExpiresIn=CONFIG["s3_expires"],
            )

            # 3. UPDATE
            cur.execute("""
                UPDATE attachment
                SET reference  = %s,
                    payload    = %s,
                    updated_ts = (EXTRACT(EPOCH FROM now()) * 1000)::bigint
                WHERE id = %s
            """, (new_presigned, json.dumps(payload, ensure_ascii=False), old_id))

            ok += 1
            if ok % 10 == 0 or ok == len(rows):
                print(f"  [OK {ok}/{len(rows)}] id={old_id}", flush=True)

        except ClientError as e:
            conn.rollback()
            fail += 1
            print(f"  [S3 FAIL] id={old_id}: {e}", flush=True)
        except Exception as e:
            conn.rollback()
            fail += 1
            print(f"  [FAIL] id={old_id}: {e}", flush=True)

    conn.commit()
    print(f"\nV done: ok={ok} fail={fail}")
    conn.close()
    return 0 if fail == 0 else 2


if __name__ == "__main__":
    sys.exit(main())

迁移后操作

归档原始 SQLite 数据库
在 PostgreSQL 上执行 VACUUM FULL 优化空间
验证迁移后的数据数量是否与原数据库一致

pgsql 部署

最后贴一个pgsql的docker-compose.yml

部署前提：

需要创建data 和backup目录。data存储数据库数据，用户官方要求为70:70 backup为备份使用

mkdir data && sudo chown -R 70:70 data && sudo chmod 700 data

创建.env 负责输入环境变量

services:
  postgres:
    image: postgres:18-alpine
    container_name: postgres
    restart: unless-stopped
    environment:
      POSTGRES_DB: ${POSTGRES_DB}
      POSTGRES_USER: ${POSTGRES_USER}
      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
      # 初始化数据库参数
      # -E UTF8
      # --data-checksums 开启数据页校验可以检测磁盘损坏导致的数据异常
      POSTGRES_INITDB_ARGS: "-E UTF8 --data-checksums"
      # 设置容器时区
      TZ: Asia/Shanghai
      # PostgreSQL内部时区
      PGTZ: Asia/Shanghai
    volumes:
      - ./data:/var/lib/postgresql
      - ./backup:/backup
    ports:
      - "5432:5432"
    command:
      - postgres
      # PostgreSQL 数据缓存，使用服务器内存作为数据库共享缓存，推荐约占物理内存 25%左右
      - -c
      - shared_buffers=512MB
      # 查询优化器估算可用缓存大小，不会实际分配内存，用于帮助优化器选择索引扫描还是全表扫描
      - -c
      - effective_cache_size=1536MB
      # 数据库维护操作使用的最大内存，用于 VACUUM、CREATE INDEX、ALTER TABLE 等操作，设置过大会影响其他连接
      - -c
      - maintenance_work_mem=128MB
      # 单个 SQL 操作可使用的工作内存
      - -c
      - work_mem=4MB
      # checkpoint 写入磁盘的平滑程度
      - -c
      - checkpoint_completion_target=0.9
      # WAL 日志缓存大小
      # WAL 用于保证事务可靠性
      # 写入数据前先写 WAL 日志
      - -c
      - wal_buffers=16MB
      # 并发 IO 数量估计
      - -c
      - effective_io_concurrency=200
      # 最大客户端连接数量
      - -c
      - max_connections=100
    healthcheck:
      test:
        [
          "CMD-SHELL",
          "pg_isready -U ${POSTGRES_USER} -d ${POSTGRES_DB}"
        ]
      # 每10秒检测一次
      interval: 10s
      # 单次检测最多等待5秒
      timeout: 5s
      # 连续失败5次认为异常
      retries: 5
      # 启动后等待30秒再检测
      start_period: 30s
    security_opt:
      # 禁止容器进程提升额外Linux权限
      # 增强容器安全性
      - no-new-privileges:true
    tmpfs:
      # 使用内存文件系统
      # 临时文件不写入磁盘容器删除自动消失
      - /tmp
      - /run
    logging:
      # Docker默认日志驱动
      driver: json-file
      # 限制日志大小
      # 防止数据库日志占满磁盘
      options:
        max-size: "20m"
        max-file: "5"

    networks:
      - database

networks:
  database:
    driver: bridge

前提条件

Docker Compose 配置

验证连接成功

数据迁移脚本

迁移脚本 (migrate.py)

S3 凭证轮换脚本

迁移后操作

pgsql 部署

迁移脚本 (`migrate.py`)