diff options
Diffstat (limited to 'api/app.py')
| -rw-r--r-- | api/app.py | 137 |
1 files changed, 4 insertions, 133 deletions
@@ -1,7 +1,6 @@ -from flask import Flask, render_template, jsonify, request +from flask import Flask, render_template, jsonify, request, redirect from flask_cors import CORS -import psycopg2 -from psycopg2 import Error +from .database import PostgresHandler import os import secrets from dotenv import load_dotenv @@ -11,130 +10,6 @@ load_dotenv() app = Flask(__name__) CORS(app) -class PostgresHandler: - def __init__(self, username: str, password: str, host_name: str, port: int, database: str): - db_params = { - "dbname": database, - "user": username, - "password": password, - "host": host_name, - "port": port - } - self._connection = psycopg2.connect(**db_params) - print("Handler Success") - - - def create_table(self, name: str, column: str): - cursor = self._connection.cursor() - cursor.execute(f"CREATE TABLE IF NOT EXISTS {name} ({column})") - self._connection.commit() - cursor.close() - - def clear_table(self, name: str): - cursor = self._connection.cursor() - cursor.execute(f"DELETE FROM {name}") - self._connection.commit() - cursor.close() - - def check_row_exists(self, table_name: str, column_name: str, value: str): - cursor = self._connection.cursor() - query = f"SELECT 1 FROM {table_name} WHERE {column_name} = %s" - cursor.execute(query, (value,)) - result = cursor.fetchone() - cursor.close() - - if result is not None: - return True - else: - return False - - def insert_row(self, table_name, column, data): - try: - cursor = self._connection.cursor() - placeholders = ', '.join(['%s'] * len(data)) - query = f"INSERT INTO {table_name} ({column}) VALUES ({placeholders})" - cursor.execute(query, data) - self._connection.commit() - print("Data Inserted:", data) - except Error as err: - self._connection.rollback() - print("Error inserting data") - print(err) - if "duplicate key" not in str(err).lower(): - return False - return True - - def get_rows(self, table_name: str, column: str, value: str): - try: - cursor = self._connection.cursor() - query = f"SELECT * FROM {table_name} WHERE {column} = %s" - cursor.execute(query, (value,)) - result = cursor.fetchall() - return result - except Error as e: - self._connection.rollback() - print(f"Failed to fetch row from {table_name} WHERE {column} is {value}") - print(e) - return False - - def get_random_row(self, table_name: str, count: int, condition: str = None): - if condition is None: - condition = "1 = 1" - try: - cursor = self._connection.cursor() - query = f"SELECT * FROM {table_name} WHERE {condition} ORDER BY RANDOM() LIMIT {str(count)}" - cursor.execute(query) - result = cursor.fetchall() - return result - except Error as e: - self._connection.rollback() - print(f"Failed to select random rows from {table_name}") - print(e) - return False - - def check_health(self): - cursor = self._connection.cursor() - cursor.execute("SELECT 1") - result = cursor.fetchone() - cursor.close() - if result is not None: - return True - else: - return False - - def delete_row(self, table_name: str, column: str, value: str): - try: - cursor = self._connection.cursor() - query = f"DELETE FROM {table_name} WHERE {column} = %s" - cursor.execute(query, (value,)) - self._connection.commit() - print("Data Deleted:", value) - except Error as e: - self._connection.rollback() - print(f"Failed to delete row from {table_name} WHERE {column} is {value}") - print(e) - return False - return True - - def get_distinct_col(self, table_name: str, column: str): - try: - cursor = self._connection.cursor() - query = f"SELECT DISTINCT {column} FROM {table_name}" - cursor.execute(query) - result = cursor.fetchall() - cursor.close() - return [row[0] for row in result] - except Error as e: - self._connection.rollback() - print(f"Failed to get unique values from {column} in {table_name}") - print(e) - return False - - - - def close_connection(self): - self._connection.close() - def create_database_connection(): """ Creates a database connection using the environment variables @@ -154,11 +29,7 @@ initialize_auth_database() @app.route('/') def index_demo(): - return render_template('index.html') - -@app.route('/server_auth') -def server_side_auth_demo(): - return render_template('server_auth.html') + return redirect("https://captcha.moekyun.me", code=302) # Demo Site @app.route('/api/list_orgs') def get_available_orgs(): @@ -239,7 +110,7 @@ def verify_answers(): def clear_sessions(): auth = request.headers.get("Authorization") cron_secret = os.environ.get("CRON_SECRET") - print("Recieved Request to Clear Session: Checking if " + auth + " matches " + cron_secret) + print("Recieved Request to Clear Session: Checking if " + str(auth) + " matches " + str(cron_secret)) if auth.strip() != cron_secret.strip(): return jsonify({"error": "Unauthorized"}), 401 server = create_database_connection() |