import os
import psycopg2
from flask import Flask, redirect, render_template_string, request
app = Flask(__name__)
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
)
"""
)
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(
"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():
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():
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)