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, unique=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_id = db.Column(db.Integer, db.ForeignKey("application.id"), 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, address, comment=""):
57
self.application_id = application.id
58
self.name = name
59
self.address = address
60
self.comment = comment
61
Base = declarative_base()
63
class Status(Base):
65
__table_args = (
66
{
67
"timescaledb_hypertable": {
68
"time_column_name": "time",
69
},
70
}
71
)
72
__tablename__ = "status"
73
id = db.Column(db.Integer, unique=True, nullable=False, autoincrement=True)
74
endpoint_id = db.Column(db.Integer, nullable=False)
75
time = db.Column(db.DateTime, index=True, default=datetime.datetime.utcnow, primary_key=True)
76
status = db.Column(db.SmallInteger, nullable=False)
78
endpoint = db.relationship("Endpoint", back_populates="statuses")
80
def __init__(self, endpoint, status):
82
self.endpoint_id = endpoint.id
83
self.status = status
84
def ping(endpoint):
87
url = endpoint.address
88
response = httpx.get(url)
89
return response.status_code
90
@app.context_processor
93
def default():
94
return {
95
"session": flask.session,
96
}
97
@app.route("/")
100
def dashboard():
101
return flask.render_template("dashboard.html", apps=Application.query.all())
102
@app.route("/login", methods=["GET"])
105
def login():
106
return flask.render_template("login.html")
107
@app.route("/signup", methods=["GET"])
110
def signup():
111
return flask.render_template("signup.html")
112
@app.route("/new-app", methods=["GET"])
115
def new_app():
116
if not flask.session.get("username"):
117
return flask.redirect("/login", code=303)
118
return flask.render_template("new-app.html")
119
@app.route("/new-app", methods=["POST"])
122
def new_app_post():
123
if not flask.session.get("username"):
124
return flask.redirect("/login", code=303)
125
if Application.query.filter_by(name=flask.request.form["name"]).first():
126
flask.flash("Application already exists")
127
return flask.redirect("/new-app", code=303)
128
new_app_ = Application(
130
flask.request.form["name"],
131
db.session.get(User, flask.session["username"]),
132
)
133
db.session.add(new_app_)
134
db.session.commit()
135
return flask.redirect("/", code=303)
136
@app.route("/login", methods=["POST"])
139
def login_post():
140
user = db.session.get(User, flask.request.form["username"])
141
if not user:
142
flask.flash("Username doesn't exist")
143
return flask.redirect("/signup", code=303)
144
if not bcrypt.check_password_hash(user.password, flask.request.form["password"]):
145
flask.flash("Wrong password")
146
return flask.redirect("/signup", code=303)
147
flask.session["username"] = user.username
149
return flask.redirect("/", code=303)
150
@app.route("/logout")
153
def logout():
154
flask.session.pop("username", None)
155
return flask.redirect("/", code=303)
156
@app.route("/signup", methods=["POST"])
159
def signup_post():
160
if flask.request.form["password"] != flask.request.form["password2"]:
161
flask.flash("Passwords do not match")
162
return flask.redirect("/signup", code=303)
163
if db.session.get(User, flask.request.form["username"]):
164
flask.flash("Username already exists")
165
return flask.redirect("/signup", code=303)
166
if len(flask.request.form["password"]) < 8:
167
flask.flash("Password must be at least 8 characters")
168
return flask.redirect("/signup", code=303)
169
if len(flask.request.form["username"]) < 4:
170
flask.flash("Username must be at least 4 characters")
171
return flask.redirect("/signup", code=303)
172
new_user = User(
174
flask.request.form["username"],
175
flask.request.form["password"],
176
)
177
db.session.add(new_user)
178
db.session.commit()
179
flask.session["username"] = new_user.username
180
return flask.redirect("/", code=303)
181
@app.route("/timeline/<endpoint_id>")
184
def info(endpoint_id):
185
return flask.render_template("timeline.html", endpoint=endpoint_id)
186
@app.route("/app/<int:app_id>/")
189
def app_info(app_id):
190
app_ = db.session.get(Application, app_id)
191
return flask.render_template("app.html", app=app_)
192
@app.route("/app/<int:app_id>/edit/")
195
def app_editor(app_id):
196
if flask.session.get("username") != db.session.get(Application, app_id).owner_name:
197
flask.abort(403)
198
app_ = db.session.get(Application, app_id)
199
return flask.render_template("app-editor.html", app=app_)
200
@app.route("/app/<int:app_id>/edit/<int:endpoint_id>", methods=["POST"])
203
def endpoint_edit(app_id, endpoint_id):
204
if flask.session.get("username") != db.session.get(Application, app_id).owner_name:
205
flask.abort(403)
206
endpoint = db.session.get(Endpoint, endpoint_id)
207
if flask.request.form.get("delete") == "delete":
208
db.session.delete(endpoint)
209
db.session.commit()
210
else:
211
endpoint.name = flask.request.form["name"]
212
endpoint.address = flask.request.form["url"]
213
endpoint.comment = flask.request.form["comment"]
214
db.session.commit()
215
return flask.redirect(f"/app/{app_id}/edit", code=303)
216
@app.route("/app/<int:app_id>/add-endpoint", methods=["POST"])
219
def app_add_endpoint(app_id):
220
if flask.session.get("username") != db.session.get(Application, app_id).owner_name:
221
flask.abort(403)
222
app_ = db.session.get(Application, app_id)
223
endpoint = Endpoint(app_,
224
flask.request.form["name"],
225
flask.request.form["url"],
226
flask.request.form["comment"])
227
db.session.add(endpoint)
228
db.session.commit()
229
return flask.redirect(f"/app/{app_id}/edit", code=303)
230