37 lines
1.4 KiB
Python
37 lines
1.4 KiB
Python
from typing import List, Optional, Any
|
|
from psycopg import AsyncConnection
|
|
|
|
|
|
class ScadaRepository:
|
|
|
|
@staticmethod
|
|
async def get_scadas(conn: AsyncConnection) -> List[dict]:
|
|
"""
|
|
查询pg数据库中,scada_info 的所有记录
|
|
:param conn: 异步数据库连接
|
|
:return: 包含所有记录的列表,每条记录为一个字典
|
|
"""
|
|
async with conn.cursor() as cur:
|
|
await cur.execute(
|
|
"""
|
|
SELECT id, type, associated_element_id, transmission_mode, transmission_frequency, reliability
|
|
FROM public.scada_info
|
|
"""
|
|
)
|
|
records = await cur.fetchall()
|
|
# 将查询结果转换为字典列表（假设 record 是字典）
|
|
scada_infos = []
|
|
for record in records:
|
|
scada_infos.append(
|
|
{
|
|
"id": record["id"], # 使用字典键
|
|
"type": record["type"],
|
|
"associated_element_id": record["associated_element_id"],
|
|
"transmission_mode": record["transmission_mode"],
|
|
"transmission_frequency": record["transmission_frequency"],
|
|
"reliability": record["reliability"],
|
|
}
|
|
)
|
|
|
|
return scada_infos
|