simple csv export for user table implemented

slaeforms/app.py

@@ -9,10 +9,11 @@ from sqlalchemy import Integer, String, Column, Float
 from datetime import datetime
 import uuid
 from sqlalchemy.exc import SQLAlchemyError
-from sqlalchemy import select
+from sqlalchemy import select, join
 from sqlalchemy.orm import DeclarativeBase
 from flask_wtf.csrf import CSRFProtect
 import os
+import csv
 
 random_order = True
 # activate environment: cd C:\Users\Jan\Google Drive\Master Stuff\Code\SLAEForms Testing\.venv\Scripts\
@@ -37,40 +38,15 @@ db.init_app(app)
 app.secret_key = b"29fe9e8edd407c5491d4f1c05632d9fa33e26ed8734a3f5e080ebac3772a555a"
 
 UPLOAD_FOLDER = 'uploads'
+EXPORT_FOLDER = 'exports'
 
 #csrf = CSRFProtect(app) #enable CSRF protection globally
 
 
-
-#---------testing JSON Stuff
-#open the json file with the config
-testconfigfile = open("singleformconfig.json", encoding='utf-8')
-#convert it to dict
-testconfig = json.load(testconfigfile)
-testconfigfile.close()
-# get the questions:    Questions is a list that contains the keys of the dictionary
-questions = list(testconfig)
-
-
-
-
-
-
 #------------------------------------------------------------------------
-# Setting up DB Models, response -> should be removed eventually and replaced by json files and automatic generation ----------------------------
+# Setting up DB Models
 
-# create the model for the response table
-class Response(db.Model):
-    id = db.Column("id",db.UUID(as_uuid=True), primary_key=True, nullable=False)
-    user_id = db.Column("user_id",db.UUID(as_uuid=True), nullable=False)
-    question_title = db.Column("question_title",db.String(30))
-    likert_result = db.Column("likert_result",db.Integer, nullable=False)
-    notes = db.Column("notes",db.String(200))
-    date_created = db.Column("date_created",db.DateTime)
-    def __repr__(self) -> str:
-        return "<Response %r>" % self.id
-
-# This table is always created
+# This table is always created, tracks all users
 class User(db.Model): 
     user_id = db.Column("user_id",db.UUID(as_uuid=True), primary_key=True, nullable=False)
     device_id = db.Column("device_id",db.UUID(as_uuid=True), nullable=False)
@@ -87,13 +63,8 @@ except SQLAlchemyError as e:
         print("Error occurred during database creation:", str(e))
 
 
-
-# -------Testing the actual final form from json------------------------------------------------
-
-
-
-
-
+#-----------------------------------------------------------------------------
+#open, parse and execute json file
 
 #open the json file with the config
 configfile = open("default.json", encoding='utf-8') #todo replace with other name
@@ -155,6 +126,10 @@ try:
 except SQLAlchemyError as e:
         print("Error occurred during database creation:", str(e))
 
+
+#------------------------------------------------------------------------------
+#actual page logic with start, form and send
+
 @app.route("/start", methods=["GET", "POST"])
 def startpage():
     session.permanent = False
@@ -413,6 +388,46 @@ def update_session():
 
 # Database stuff------------------------------------------------------------------------------
 
+def create_csv(model, filename):
+    filename = filename + ".csv"
+    # Query all data from the table
+    data = db.session.query(model).all()
+
+    # Get the column names from the model
+    column_names = [column.name for column in model.__table__.columns]
+
+    # Open a CSV file and write data
+    path = os.path.join(EXPORT_FOLDER, filename)
+    with open(path, 'w', newline='') as file:
+        writer = csv.writer(file)
+        writer.writerow(column_names)  # Write header
+        for row in data:
+            writer.writerow([getattr(row, column) for column in column_names])
+
+
+# export CSV
+@app.route("/export_csv")
+def export_csv():
+
+    create_csv(User, "usertable")
+
+    """
+    meta = db.metadata
+    tables = meta.tables.keys()
+    print("tables: ",tables)
+    print("testquerys:")
+    qtable = meta.tables["default_demographic_test"]
+    query1 = select(qtable).where(qtable.c.alter == 78)
+    print("Query 1: ", query1)
+    print("Query 1 result: ")
+    result = db.session.execute(query1)
+    print("Columns: ", result.columns)
+    for row in result:
+        print(row)
+    """
+    
+    return redirect("/")
+
 # the contents of all tables
 @app.route("/table_contents")
 def table_contents():
@@ -464,6 +479,20 @@ def show_tables():
     tables = meta.tables
     return render_template('show_tables.html', tables=tables)
 
+# Control Panel ---------------------------------------------------------
+
+@app.route("/upload_configs")
+def upload_configs():
+    links = []
+    for rule in app.url_map.iter_rules():
+        # Filter out rules we can't navigate to in a browser
+        # and rules that require parameters
+        if "GET" in rule.methods and has_no_empty_params(rule):
+            url = url_for(rule.endpoint, **(rule.defaults or {}))
+            links.append((url, rule.endpoint))
+    return render_template("all_links.html", links=links)
+
+
 # Root page -----------------------------
 
 def has_no_empty_params(rule):
@@ -526,4 +555,4 @@ def delete_all_entries():
         db.session.close()
 
 if __name__ == '__main__':
-    app.run(debug=True)
+    app.run()

slaeforms/exports/usertable.csv

@@ -0,0 +1,2 @@
+user_id,device_id,question_order,date_created
+662dc5a5-ff08-4270-ad33-444e8dd01514,50bc84f6-6c55-4277-9651-5c3c98e973f5,"{'Block 0': [('video_1', 'video_1'), ('video_2', 'video_2'), ('video_3', 'video_3')], 'Block 1': ['empty_stimulus'], 'Block 2': ['video_1', 'video_3', 'video_2']}",2024-06-25 12:01:22.516505