feat(store): add auto_ingest config + pg connect/statement timeouts
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -70,14 +70,14 @@ anneng:
|
||||
app_path: 'D:\SoftWare\SoftWare Installation\@ane-electron-uiapp\安能全网门户.exe'
|
||||
|
||||
# ----------------------------------------------------------------------------
|
||||
# PostgreSQL 数据持久化(到货核销数据入库,详见 db_store.py)
|
||||
# PostgreSQL 数据持久化(到货核销数据入库,详见 store.py)
|
||||
# ----------------------------------------------------------------------------
|
||||
# createdb 会连接名为 postgres 的维护库来创建下方 dbname 指定的数据库。
|
||||
# 命令行:
|
||||
# python db_store.py createdb 创建数据库(幂等)
|
||||
# python db_store.py init 建表(幂等)
|
||||
# python db_store.py ingest [site] 入库全站或单站(幂等 UPSERT)
|
||||
# python db_store.py all createdb → init → 全站 ingest
|
||||
# python -m inbound_verify.store createdb 创建数据库(幂等)
|
||||
# python -m inbound_verify.store init 建表(幂等)
|
||||
# python -m inbound_verify.store ingest [site] 入库全站或单站(幂等 UPSERT)
|
||||
# python -m inbound_verify.store all createdb → init → 全站 ingest
|
||||
postgres:
|
||||
host: 127.0.0.1
|
||||
port: 5432
|
||||
@@ -87,3 +87,7 @@ postgres:
|
||||
dbname: CQHXDB
|
||||
# 承载到货核销表的专用 schema(隔离 public);表建在此 schema 下。
|
||||
schema: inbound_verify
|
||||
# 下载成功后自动入库(钩子,见 runtime._persist_to_db);false=跳过(无 PG/cpolar 的开发机)。
|
||||
auto_ingest: true
|
||||
# PG 连接超时(秒);cpolar 抖动时快速失败,不拖垮下载 worker。
|
||||
connect_timeout_seconds: 5
|
||||
|
||||
@@ -16,6 +16,7 @@ store.py — 到货核销数据持久化(PostgreSQL)
|
||||
python -m inbound_verify.store createdb 创建数据库(幂等)
|
||||
python -m inbound_verify.store init 建表(幂等 CREATE TABLE IF NOT EXISTS)
|
||||
python -m inbound_verify.store ingest [site] 入库全站或单站(幂等 UPSERT)
|
||||
python -m inbound_verify.store ingest-one <site> <kind> 仅入库指定站/类(钩子同款路由)
|
||||
python -m inbound_verify.store all createdb → init → 全站 ingest 一条龙
|
||||
"""
|
||||
|
||||
@@ -61,12 +62,15 @@ def _load_pg_config():
|
||||
"password": pg.get("password", ""),
|
||||
"dbname": pg.get("dbname", "CQHXDB"),
|
||||
"schema": pg.get("schema", "inbound_verify"),
|
||||
"auto_ingest": bool(pg.get("auto_ingest", True)),
|
||||
"connect_timeout_seconds": int(pg.get("connect_timeout_seconds", 5)),
|
||||
}
|
||||
|
||||
|
||||
def _connect(dbname):
|
||||
"""用关键字参数连接(避开 conninfo 对密码特殊字符的解析)。
|
||||
options 设 search_path 到专用 schema,使无 schema 限定的表名解析到该 schema。"""
|
||||
options 设 search_path 到专用 schema + 会话级 statement_timeout=30s
|
||||
(cpolar 隧道上防失控查询;connect_timeout 守连接阶段)。"""
|
||||
c = _load_pg_config()
|
||||
return psycopg.connect(
|
||||
host=c["host"],
|
||||
@@ -74,10 +78,17 @@ def _connect(dbname):
|
||||
dbname=dbname,
|
||||
user=c["user"],
|
||||
password=c["password"],
|
||||
options=f"-c search_path={c['schema']}",
|
||||
options=f"-c search_path={c['schema']} -c statement_timeout=30s",
|
||||
connect_timeout=c["connect_timeout_seconds"],
|
||||
)
|
||||
|
||||
|
||||
def ingest_enabled():
|
||||
"""是否启用下载后自动入库(config.yaml postgres.auto_ingest,默认 True)。
|
||||
供 runtime 钩子判定开关,避免它伸手进 _load_pg_config。"""
|
||||
return _load_pg_config()["auto_ingest"]
|
||||
|
||||
|
||||
# ============================== 建库 / 建表 ==============================
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user