generated from nighthawkcoders/flocker_backend
-
Notifications
You must be signed in to change notification settings - Fork 0
/
tempCodeRunnerFile.py
236 lines (204 loc) · 8.19 KB
/
tempCodeRunnerFile.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
# imports from flask
import json
import os
from urllib.parse import urljoin, urlparse
from flask import abort, redirect, render_template, request, send_from_directory, url_for, jsonify # import render_template from "public" flask libraries
from flask_login import current_user, login_user, logout_user
from flask.cli import AppGroup
from flask_login import current_user, login_required
from flask import current_app
from werkzeug.security import generate_password_hash
import shutil
# import "objects" from "this" project
from __init__ import app, db, login_manager # Key Flask objects
# API endpoints
from api.user import user_api
from api.pfp import pfp_api
from api.nestImg import nestImg_api # Justin added this, custom format for his website
from api.post import post_api
from api.channel import channel_api
from api.group import group_api
from api.section import section_api
from api.student import student_api
from api.nestPost import nestPost_api # Justin added this, custom format for his website
from api.messages_api import messages_api # Adi added this, messages for his website
from api.carphoto import car_api
from api.carChat import car_chat_api
from api.bookreview import bookreview_api
from api.vote import vote_api
# database Initialization functions
from model.carChat import CarChat
from model.user import User, initUsers
from model.section import Section, initSections
from model.group import Group, initGroups
from model.channel import Channel, initChannels
from model.post import Post, initPosts
from model.nestPost import NestPost, initNestPosts # Justin added this, custom format for his website
from model.vote import Vote, initVotes
# server only Views
# register URIs for api endpoints
app.register_blueprint(messages_api) # Adi added this, messages for his website
app.register_blueprint(user_api)
app.register_blueprint(pfp_api)
app.register_blueprint(post_api)
app.register_blueprint(channel_api)
app.register_blueprint(group_api)
app.register_blueprint(section_api)
app.register_blueprint(student_api)
app.register_blueprint(car_chat_api)
app.register_blueprint(bookreview_api)
# Added new files to create nestPosts, uses a different format than Mortensen and didn't want to touch his junk
app.register_blueprint(nestPost_api)
app.register_blueprint(nestImg_api)
app.register_blueprint(vote_api)
app.register_blueprint(car_api)
# Tell Flask-Login the view function name of your login route
login_manager.login_view = "login"
@login_manager.unauthorized_handler
def unauthorized_callback():
return redirect(url_for('login', next=request.path))
# register URIs for server pages
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
@app.context_processor
def inject_user():
return dict(current_user=current_user)
# Helper function to check if the URL is safe for redirects
def is_safe_url(target):
ref_url = urlparse(request.host_url)
test_url = urlparse(urljoin(request.host_url, target))
return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc
@app.route('/login', methods=['GET', 'POST'])
def login():
error = None
next_page = request.args.get('next', '') or request.form.get('next', '')
if request.method == 'POST':
user = User.query.filter_by(_uid=request.form['username']).first()
if user and user.is_password(request.form['password']):
login_user(user)
if not is_safe_url(next_page):
return abort(400)
return redirect(next_page or url_for('index'))
else:
error = 'Invalid username or password.'
return render_template("login.html", error=error, next=next_page)
@app.route('/logout')
def logout():
logout_user()
return redirect(url_for('index'))
@app.errorhandler(404) # catch for URL not found
def page_not_found(e):
# note that we set the 404 status explicitly
return render_template('404.html'), 404
@app.route('/') # connects default URL to index() function
def index():
print("Home:", current_user)
return render_template("index.html")
@app.route('/users/table')
@login_required
def utable():
users = User.query.all()
return render_template("utable.html", user_data=users)
@app.route('/users/table2')
@login_required
def u2table():
users = User.query.all()
return render_template("u2table.html", user_data=users)
# Helper function to extract uploads for a user (ie PFP image)
@app.route('/uploads/<path:filename>')
def uploaded_file(filename):
return send_from_directory(current_app.config['UPLOAD_FOLDER'], filename)
@app.route('/users/delete/<int:user_id>', methods=['DELETE'])
@login_required
def delete_user(user_id):
user = User.query.get(user_id)
if user:
user.delete()
return jsonify({'message': 'User deleted successfully'}), 200
return jsonify({'error': 'User not found'}), 404
@app.route('/users/reset_password/<int:user_id>', methods=['POST'])
@login_required
def reset_password(user_id):
if current_user.role != 'Admin':
return jsonify({'error': 'Unauthorized'}), 403
user = User.query.get(user_id)
if not user:
return jsonify({'error': 'User not found'}), 404
# Set the new password
if user.update({"password": app.config['DEFAULT_PASSWORD']}):
return jsonify({'message': 'Password reset successfully'}), 200
return jsonify({'error': 'Password reset failed'}), 500
# Create an AppGroup for custom commands
custom_cli = AppGroup('custom', help='Custom commands')
# Define a command to run the data generation functions
@custom_cli.command('generate_data')
def generate_data():
initUsers()
initSections()
initGroups()
initChannels()
initPosts()
initNestPosts()
initVotes()
# Backup the old database
def backup_database(db_uri, backup_uri):
"""Backup the current database."""
if backup_uri:
db_path = db_uri.replace('sqlite:///', 'instance/')
backup_path = backup_uri.replace('sqlite:///', 'instance/')
shutil.copyfile(db_path, backup_path)
print(f"Database backed up to {backup_path}")
else:
print("Backup not supported for production database.")
# Extract data from the existing database
def extract_data():
data = {}
with app.app_context():
data['users'] = [user.read() for user in User.query.all()]
data['sections'] = [section.read() for section in Section.query.all()]
data['groups'] = [group.read() for group in Group.query.all()]
data['channels'] = [channel.read() for channel in Channel.query.all()]
data['posts'] = [post.read() for post in Post.query.all()]
return data
# Save extracted data to JSON files
def save_data_to_json(data, directory='backup'):
if not os.path.exists(directory):
os.makedirs(directory)
for table, records in data.items():
with open(os.path.join(directory, f'{table}.json'), 'w') as f:
json.dump(records, f)
print(f"Data backed up to {directory} directory.")
# Load data from JSON files
def load_data_from_json(directory='backup'):
data = {}
for table in ['users', 'sections', 'groups', 'channels', 'posts']:
with open(os.path.join(directory, f'{table}.json'), 'r') as f:
data[table] = json.load(f)
return data
# Restore data to the new database
def restore_data(data):
with app.app_context():
users = User.restore(data['users'])
_ = Section.restore(data['sections'])
_ = Group.restore(data['groups'], users)
_ = Channel.restore(data['channels'])
_ = Post.restore(data['posts'])
print("Data restored to the new database.")
# Define a command to backup data
@custom_cli.command('backup_data')
def backup_data():
data = extract_data()
save_data_to_json(data)
backup_database(app.config['SQLALCHEMY_DATABASE_URI'], app.config['SQLALCHEMY_BACKUP_URI'])
# Define a command to restore data
@custom_cli.command('restore_data')
def restore_data_command():
data = load_data_from_json()
restore_data(data)
# Register the custom command group with the Flask application
app.cli.add_command(custom_cli)
# this runs the flask application on the development server
if __name__ == "__main__":
# change name for testing
app.run(debug=True, host="0.0.0.0", port="8887")