OrgTJWater/TJWaterServerBinary
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改scheme.py的批量存储方法

Browse Source
This commit is contained in:
Jiang
2026-01-07 18:09:22 +08:00
parent 73f14c6f83
commit 87c5a07219
1 changed files with 20 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
timescaledb/schemas/scheme.py
View File

@@ -18,18 +18,18 @@ class SchemeRepository:
if not data: if not data:
return return
# 假设同一批次的数据时间、scheme_type、scheme_name 是相同的 # 获取批次中所有不同的时间点
target_time = data[0]["time"] all_times = list(set(item["time"] for item in data))
target_scheme_type = data[0]["scheme_type"] target_scheme_type = data[0]["scheme_type"]
target_scheme_name = data[0]["scheme_name"] target_scheme_name = data[0]["scheme_name"]
# 使用事务确保原子性 # 使用事务确保原子性
async with conn.transaction(): async with conn.transaction():
async with conn.cursor() as cur: async with conn.cursor() as cur:
# 1. 先删除该时间点、scheme_type、scheme_name 的旧数据 # 1. 删除该批次涉及的所有时间点、scheme_type、scheme_name 的旧数据
await cur.execute( await cur.execute(
"DELETE FROM scheme.link_simulation WHERE time = %s AND scheme_type = %s AND scheme_name = %s", "DELETE FROM scheme.link_simulation WHERE time = ANY(%s) AND scheme_type = %s AND scheme_name = %s",
(target_time, target_scheme_type, target_scheme_name), (all_times, target_scheme_type, target_scheme_name),
) )
# 2. 使用 COPY 快速写入新数据 # 2. 使用 COPY 快速写入新数据
@@ -60,18 +60,18 @@ class SchemeRepository:
if not data: if not data:
return return
# 假设同一批次的数据时间、scheme_type、scheme_name 是相同的 # 获取批次中所有不同的时间点
target_time = data[0]["time"] all_times = list(set(item["time"] for item in data))
target_scheme_type = data[0]["scheme_type"] target_scheme_type = data[0]["scheme_type"]
target_scheme_name = data[0]["scheme_name"] target_scheme_name = data[0]["scheme_name"]
# 使用事务确保原子性 # 使用事务确保原子性
with conn.transaction(): with conn.transaction():
with conn.cursor() as cur: with conn.cursor() as cur:
# 1. 先删除该时间点、scheme_type、scheme_name 的旧数据 # 1. 删除该批次涉及的所有时间点、scheme_type、scheme_name 的旧数据
cur.execute( cur.execute(
"DELETE FROM scheme.link_simulation WHERE time = %s AND scheme_type = %s AND scheme_name = %s", "DELETE FROM scheme.link_simulation WHERE time = ANY(%s) AND scheme_type = %s AND scheme_name = %s",
(target_time, target_scheme_type, target_scheme_name), (all_times, target_scheme_type, target_scheme_name),
) )
# 2. 使用 COPY 快速写入新数据 # 2. 使用 COPY 快速写入新数据
@@ -252,18 +252,18 @@ class SchemeRepository:
if not data: if not data:
return return
# 假设同一批次的数据时间、scheme_type、scheme_name 是相同的 # 获取批次中所有不同的时间点
target_time = data[0]["time"] all_times = list(set(item["time"] for item in data))
target_scheme_type = data[0]["scheme_type"] target_scheme_type = data[0]["scheme_type"]
target_scheme_name = data[0]["scheme_name"] target_scheme_name = data[0]["scheme_name"]
# 使用事务确保原子性 # 使用事务确保原子性
async with conn.transaction(): async with conn.transaction():
async with conn.cursor() as cur: async with conn.cursor() as cur:
# 1. 先删除该时间点、scheme_type、scheme_name 的旧数据 # 1. 删除该批次涉及的所有时间点、scheme_type、scheme_name 的旧数据
await cur.execute( await cur.execute(
"DELETE FROM scheme.node_simulation WHERE time = %s AND scheme_type = %s AND scheme_name = %s", "DELETE FROM scheme.node_simulation WHERE time = ANY(%s) AND scheme_type = %s AND scheme_name = %s",
(target_time, target_scheme_type, target_scheme_name), (all_times, target_scheme_type, target_scheme_name),
) )
# 2. 使用 COPY 快速写入新数据 # 2. 使用 COPY 快速写入新数据
@@ -289,18 +289,18 @@ class SchemeRepository:
if not data: if not data:
return return
# 假设同一批次的数据时间、scheme_type、scheme_name 是相同的 # 获取批次中所有不同的时间点
target_time = data[0]["time"] all_times = list(set(item["time"] for item in data))
target_scheme_type = data[0]["scheme_type"] target_scheme_type = data[0]["scheme_type"]
target_scheme_name = data[0]["scheme_name"] target_scheme_name = data[0]["scheme_name"]
# 使用事务确保原子性 # 使用事务确保原子性
with conn.transaction(): with conn.transaction():
with conn.cursor() as cur: with conn.cursor() as cur:
# 1. 先删除该时间点、scheme_type、scheme_name 的旧数据 # 1. 删除该批次涉及的所有时间点、scheme_type、scheme_name 的旧数据
cur.execute( cur.execute(
"DELETE FROM scheme.node_simulation WHERE time = %s AND scheme_type = %s AND scheme_name = %s", "DELETE FROM scheme.node_simulation WHERE time = ANY(%s) AND scheme_type = %s AND scheme_name = %s",
(target_time, target_scheme_type, target_scheme_name), (all_times, target_scheme_type, target_scheme_name),
) )
# 2. 使用 COPY 快速写入新数据 # 2. 使用 COPY 快速写入新数据