Python script, ASCII text executable
1
import datetime
2
import flask
4
from flask_sqlalchemy import SQLAlchemy
5
from flask_bcrypt import Bcrypt
6
from flask_migrate import Migrate
7
from sqlalchemy.orm import declarative_base
9
import httpx
11
app = flask.Flask(__name__)
13
app.config["SQLALCHEMY_DATABASE_URI"] = \
14
"postgresql://echo:1234@localhost:5432/echo"
15
db = SQLAlchemy(app)
16
bcrypt = Bcrypt(app)
17
migrate = Migrate(app, db)
18
app.config["SESSION_TYPE"] = "filesystem"
19
app.config["SECRET_KEY"] = "super secret"
20
with app.app_context():
22
class User(db.Model):
23
username = db.Column(db.String(64), unique=True, nullable=False, primary_key=True)
24
password = db.Column(db.String(72), nullable=False)
25
admin = db.Column(db.Boolean, nullable=False, default=False)
26
applications = db.relationship("Application", back_populates="owner")
28
def __init__(self, username, password, admin=False):
30
self.username = username
31
self.password = bcrypt.generate_password_hash(password).decode("utf-8")
32
self.admin = admin
33
class Application(db.Model):
35
id = db.Column(db.Integer, primary_key=True, autoincrement=True, default=0)
36
name = db.Column(db.String(64), unique=True, nullable=False)
37
owner_name = db.Column(db.String(64), db.ForeignKey("user.username"), nullable=False)
38
owner = db.relationship("User", back_populates="applications")
40
endpoints = db.relationship("Endpoint", back_populates="application")
42
def __init__(self, name, owner):
44
self.name = name
45
self.owner_name = owner.username
46
class Endpoint(db.Model):
48
id = db.Column(db.Integer, unique=True, nullable=False, primary_key=True, autoincrement=True)
49
application_name = db.Column(db.String(64), db.ForeignKey("application.name"), nullable=False)
50
address = db.Column(db.String(2048), nullable=False)
51
name = db.Column(db.String(64), nullable=False)
52
comment = db.Column(db.String(2048), nullable=True)
53
application = db.relationship("Application", back_populates="endpoints")
55
def __init__(self, application_name):
57
self.application_name = application_name
58
Base = declarative_base()
60
class Status(Base):
62
__table_args = (
63
{
64
"timescaledb_hypertable": {
65
"time_column_name": "time",
66
},
67
}
68
)
69
__tablename__ = "status"
70
id = db.Column(db.Integer, unique=True, nullable=False, autoincrement=True)
71
endpoint_id = db.Column(db.Integer, nullable=False)
72
time = db.Column(db.DateTime, index=True, default=datetime.datetime.utcnow, primary_key=True)
73
status = db.Column(db.SmallInteger, nullable=False)
75
endpoint = db.relationship("Endpoint", back_populates="statuses")
77
def __init__(self, endpoint, status):
79
self.endpoint_id = endpoint.id
80
self.status = status
81
def ping(endpoint):
84
url = endpoint.address
85
response = httpx.get(url)
86
return response.status_code
87
@app.context_processor
90
def default():
91
return {
92
"session": flask.session,
93
}
94
@app.route("/")
97
def dashboard():
98
return flask.render_template("dashboard.html", apps=Application.query.all())
99
@app.route("/login", methods=["GET"])
102
def login():
103
return flask.render_template("login.html")
104
@app.route("/signup", methods=["GET"])
107
def signup():
108
return flask.render_template("signup.html")
109
@app.route("/new-app", methods=["GET"])
112
def new_app():
113
if not flask.session.get("username"):
114
return flask.redirect("/login", code=303)
115
return flask.render_template("new-app.html")
116
@app.route("/new-app", methods=["POST"])
119
def new_app_post():
120
if not flask.session.get("username"):
121
return flask.redirect("/login", code=303)
122
if Application.query.filter_by(name=flask.request.form["name"]).first():
123
flask.flash("Application already exists")
124
return flask.redirect("/new-app", code=303)
125
new_app_ = Application(
127
flask.request.form["name"],
128
db.session.get(User, flask.session["username"]),
129
)
130
db.session.add(new_app_)
131
db.session.commit()
132
return flask.redirect("/", code=303)
133
@app.route("/login", methods=["POST"])
136
def login_post():
137
user = db.session.get(User, flask.request.form["username"])
138
if not user:
139
flask.flash("Username doesn't exist")
140
return flask.redirect("/signup", code=303)
141
if not bcrypt.check_password_hash(user.password, flask.request.form["password"]):
142
flask.flash("Wrong password")
143
return flask.redirect("/signup", code=303)
144
flask.session["username"] = user.username
146
return flask.redirect("/", code=303)
147
@app.route("/logout")
150
def logout():
151
flask.session.pop("username", None)
152
return flask.redirect("/", code=303)
153
@app.route("/signup", methods=["POST"])
156
def signup_post():
157
if flask.request.form["password"] != flask.request.form["password2"]:
158
flask.flash("Passwords do not match")
159
return flask.redirect("/signup", code=303)
160
if db.session.get(User, flask.request.form["username"]):
161
flask.flash("Username already exists")
162
return flask.redirect("/signup", code=303)
163
if len(flask.request.form["password"]) < 8:
164
flask.flash("Password must be at least 8 characters")
165
return flask.redirect("/signup", code=303)
166
if len(flask.request.form["username"]) < 4:
167
flask.flash("Username must be at least 4 characters")
168
return flask.redirect("/signup", code=303)
169
new_user = User(
171
flask.request.form["username"],
172
flask.request.form["password"],
173
)
174
db.session.add(new_user)
175
db.session.commit()
176
flask.session["username"] = new_user.username
177
return flask.redirect("/", code=303)
178
@app.route("/timeline/<endpoint_id>")
181
def info(endpoint_id):
182
return flask.render_template("timeline.html", endpoint=endpoint_id)
183
@app.route("/app/<int:app_id>")
186
def app_info(app_id):
187
app_ = db.session.get(Application, app_id)
188
return flask.render_template("app.html", app=app_)
189
@app.route("/app/<int:app_id>/edit")
192
def app_editor(app_id):
193
if flask.session.get("username") != db.session.get(Application, app_id).owner_name:
194
flask.abort(403)
195
app_ = db.session.get(Application, app_id)
196
return flask.render_template("app-editor.html", app=app_)
197