import datetime
import os
import psycopg2
from flask import Flask, redirect, render_template_string, request
app = Flask(__name__)
# Соединение с Postgres, DATABASE_URL имеет приоритет.
def get_conn():
db_url = os.getenv("DATABASE_URL")
if db_url:
return psycopg2.connect(db_url)
return psycopg2.connect(
host=os.getenv("PGHOST"),
port=os.getenv("PGPORT", "5432"),
user=os.getenv("PGUSER"),
password=os.getenv("PGPASSWORD"),
dbname=os.getenv("PGDATABASE", "postgres"),
sslmode=os.getenv("PGSSLMODE", "require"),
)
# Гарантируем наличие таблицы для демо.
def ensure_table():
with get_conn() as conn, conn.cursor() as cur:
cur.execute(
"""
CREATE TABLE IF NOT EXISTS nubes_test_table (
id SERIAL PRIMARY KEY,
test_data TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
# HTML-шаблон интерфейса.
PAGE = """
Flask + Postgres Demo
Flask + Postgres Demo
{% if error %}
{{ error }}
{% endif %}
ID
Содержимое
Действия
{% for row in rows %}
{{ row[0] }}
{% endfor %}
"""
@app.route("/", methods=["GET", "POST"])
def index():
ensure_table()
error = None
if request.method == "POST":
content = request.form.get("txt_content") or "Это Flask сделал"
try:
with get_conn() as conn, conn.cursor() as cur:
# Проверка дубликатов: не вставляем повтор за короткое время.
cur.execute(
"SELECT test_data, created_at FROM nubes_test_table ORDER BY id DESC LIMIT 1"
)
last = cur.fetchone()
is_duplicate = False
if last:
last_text, last_created = last
now = (
datetime.datetime.now(last_created.tzinfo)
if isinstance(last_created, datetime.datetime)
else datetime.datetime.utcnow()
)
is_duplicate = (
last_text == content
and (now - last_created).total_seconds() < 3
)
if not is_duplicate:
cur.execute(
"INSERT INTO nubes_test_table (test_data) VALUES (%s)",
(content,),
)
return redirect("/")
except Exception as exc:
error = str(exc)
with get_conn() as conn, conn.cursor() as cur:
cur.execute(
"SELECT id, test_data FROM nubes_test_table ORDER BY id DESC LIMIT 20"
)
rows = cur.fetchall()
return render_template_string(PAGE, rows=rows, error=error)
@app.post("/update")
def update():
# Обновление записи по id.
with get_conn() as conn, conn.cursor() as cur:
cur.execute(
"UPDATE nubes_test_table SET test_data=%s WHERE id=%s",
(request.form.get("txt_content"), request.form.get("id")),
)
return redirect("/")
@app.post("/delete")
def delete():
# Удаление записи по id.
with get_conn() as conn, conn.cursor() as cur:
cur.execute("DELETE FROM nubes_test_table WHERE id=%s", (request.form.get("id"),))
return redirect("/")
if __name__ == "__main__":
# Локальный запуск для отладки.
app.run(debug=True, host="0.0.0.0", port=5000)
