39 lines
1.3 KiB
Python
39 lines
1.3 KiB
Python
# 2026-03-19
|
|
# pg_stats.py — тестовая функция (Test 7): возвращает агрегированную статистику
|
|
# по таблице terraform_demo_table: кол-во строк, дата первой и последней записи.
|
|
# Создаётся и удаляется в рамках тестового прогона.
|
|
#
|
|
# Entrypoint: pg_stats.get_stats
|
|
|
|
import os
|
|
import psycopg2
|
|
import json
|
|
|
|
_CODE_VERSION = "v1-test7"
|
|
|
|
|
|
def get_stats(event):
|
|
conn = psycopg2.connect(
|
|
host=os.environ["PGHOST"],
|
|
port=int(os.environ.get("PGPORT", "5432")),
|
|
dbname=os.environ["PGDATABASE"],
|
|
user=os.environ["PGUSER"],
|
|
password=os.environ["PGPASSWORD"],
|
|
sslmode=os.environ.get("PGSSLMODE", "require"),
|
|
)
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT COUNT(*) AS cnt, MIN(created_at) AS first, MAX(created_at) AS last "
|
|
"FROM terraform_demo_table"
|
|
)
|
|
row = cur.fetchone()
|
|
return {
|
|
"version": _CODE_VERSION,
|
|
"total_rows": row[0],
|
|
"first_row_at": str(row[1]) if row[1] else None,
|
|
"last_row_at": str(row[2]) if row[2] else None,
|
|
}
|
|
finally:
|
|
conn.close()
|