Background:
I built an app in which, as soon as a user browse a file and clicks on the "upload" button, a behind-the-scenes calculation takes place, which can take long minutes, and at the end the output is uploaded to GCP. I want the calculation (after clicking) not to stuck the app and that in the meantime the user will be moved to another page. At the end of the process he will receive a success/failure email.
To achieve this goal I try to use multiprocessing.
I'm having much trouble trying to turn it to multi-processed app - I keep gettin the following error message when I'm trying to start a process:
_pickle.PicklingError: Can't pickle <class 'wtforms.form.Meta'>: attribute lookup Meta on wtforms.form failed
app.py - The main function is `upload_file:
@app.route('/')
def home():
return redirect(url_for('upload_file', page_num=1))
@app.route('/<int:page_num>', methods=['GET', 'POST'])
def upload_file(page_num=1):
form = CreateXslxForm()
traffic_data = get_page_of_traffic_data(page_num)
queue = Queue()
p = Process(target=main_func, args=(queue, form))
p.start()
proc_ret = queue.get()
if proc_ret:
upload_success(*proc_ret)
else:
return render_template(constants.GeneralConstants.LANDING_PAGE, title='URL Comparison', form=form,
traffic_data=traffic_data)
def get_page_of_traffic_data(page_num):
traffic_data = urls_comparison_users_traffic.query.paginate(per_page=5, page=int(page_num), error_out=True)
return traffic_data
def upload_success(link, traffic_row, blob_name):
add_data_to_db(traffic_row)
return render_template('upload_success.html', title='Upload File', file_path=link, filename=blob_name)
def add_data_to_db(traffic_row):
# MANUAL PRE PING
try:
db.session.execute("SELECT 1;")
db.session.commit()
except:
db.session.rollback()
finally:
db.session.close()
# SESSION COMMIT, ROLLBACK, CLOSE
try:
db.session.add(traffic_row)
db.session.commit()
send_response_mail_to_user(traffic_row)
except Exception as e:
db.session.rollback()
raise e
finally:
db.session.close()
def send_response_mail_to_user(traffic_data):
sendgrid_obj = sendgrid_handler.SendgridHandler(sendgrid_key=SENDGRID_KEY)
sendgrid_obj.mails_to_send = URLComparisonEmailBuilder.build_mails(traffic_data)
sendgrid_obj.send()
if __name__ == "__main__":
app.run(port=5000, debug=True, threaded=True)
main.py
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = r'gcp-cre.json'
storage_client = storage.Client()
def get_file_from_form(form):
# form = forms.CreateXslxForm()
timestr = time.strftime(constants.DatesFormats.DATETIME)
custom_name = f"{constants.GeneralConstants.DEST_DEFAULT_NAME}-{timestr}"
# traffic_data = get_page_of_traffic_data(page_num)
if form.validate_on_submit():
load_filename = secure_filename(form.input_file.data.filename)
valid_file_name = get_valid_filename(request.form["filename"])
if valid_file_name:
custom_name = f"{valid_file_name}"
url_comp_obj = URLCompare()
result = url_comp_obj.compare_all_urls(form.input_file.data)
row_data = {
"upload_file": form.input_file.data.filename,
"output_file": custom_name,
"gcp_link": None,
"user": 'nki-tov'
}
return custom_name, result, constants.GeneralConstants.BUCKET_NAME, row_data
def get_valid_filename(s):
s = s.strip().replace(' ', '_')
return re.sub(r'(?u)[^-\w]', '', s)
def upload_to_bucket(blob_name, file, bucket_name, row_data):
'''
Upload file to a bucket
: blob_name (str) - object name
: file_path (str)
: bucket_name (str)
'''
bucket = storage_client.get_bucket(bucket_name)
blob = bucket.blob(blob_name)
blob.upload_from_file(file)
link = f'https://console.cloud.google.com/storage/browser/_details/{bucket_name}/{blob_name};tab=live_object?authuser=0'
row_data["gcp_link"] = link
traffic_row = urls_comparison_users_traffic(**row_data)
return link, traffic_row, blob_name
def main_func(queue: Queue, form):
res = get_file_from_form(form)
if res:
queue.put(upload_to_bucket(*res))
else:
queue.put(res) # None
What am I doing wrong? Is it something to do with the architecture?