"""
Contains objects that define UI related workers that perform tasks in the
background.
Copyright (c) 2015-2022 Nicholas H.Tollervey and others (see the AUTHORS file).
This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program. If not, see .
"""
import logging
import requests
from PyQt5.QtCore import pyqtSignal, QObject
logger = logging.getLogger(__name__)
WSGI = """# This file contains the WSGI configuration required to serve up your
# web application at http://.pythonanywhere.com/
# It works by setting the variable 'application' to a WSGI handler of some
# description.
#
# The below has been auto-generated BY MU for your Flask project
import sys
# add your project directory to the sys.path
project_home = '/home/{username}/{app_name}'
if project_home not in sys.path:
sys.path = [project_home] + sys.path
# import flask app but need to call it "application" for WSGI to work
from {app_name} import app as application # noqa
"""
class PythonAnywhereWorker(QObject):
"""
Encapsulates deployment, configuration and restart of a website on
PythonAnywhere.
"""
finished = pyqtSignal(str) # Emit domain when successfully finished.
error = pyqtSignal(str) # Emitted with an error description if failed.
def __init__(self, instance, username, token, files, app_name, progress):
super().__init__(None)
self.instance = instance
self.username = username # PythonAnywhere username.
self.token = token # PythonAnywhere API token.
self.files = files # {"RemotePath": "LocalPath", ... } files dict.
self.app_name = app_name # Python module containing Flask app.
self.progress = progress # Progress modal to update.
# The root API URL. The instance should be either "www" (main
# PythonAnywhere instance) or "eu" (hosted in Europe). This is
# configured in the admin widget.
self.url = (
"https://{instance}.pythonanywhere.com/api/v0/user/{username}/"
).format(instance=instance, username=username)
# Path to where web application's files are to be uploaded.
self.files_path = "files/path/home/{username}/{app_name}/".format(
username=username, app_name=app_name
)
# Path to the WSGI configuration file for the application.
self.wsgi_path = (
"/files/path/var/www/{username}_pythonanywhere_com_wsgi.py"
).format(username=username)
# Path to where the web application's static files will be found.
self.static_path = "/home/{username}/{app_name}/static/".format(
username=username, app_name=app_name
)
# The content of the WSGI file to upload.
self.wsgi_config = WSGI.format(
username=self.username, app_name=self.app_name
)
def run(self):
logger.info(
"Deploying to PythonAnywhere {instance} for: {app_name}".format(
instance=self.instance, app_name=self.app_name
)
)
headers = {"Authorization": "Token {token}".format(token=self.token)}
domain = "{username}.pythonanywhere.com".format(username=self.username)
if self.instance == "eu":
domain = "{username}.eu.pythonanywhere.com".format(
username=self.username
)
# Progress steps are calculated as the number of files to upload plus
# the five-ish additional API calls needed to configure and restart
# the web application.
self.progress.setMaximum(len(self.files) + 5)
# The counter tracks how much progress has been made.
counter = 1
try:
# Get web application details (if required).
self.progress.setValue(counter)
path = self.url + "webapps/"
response = requests.get(path, headers=headers)
response.raise_for_status()
apps = response.json()
# Create the application if it doesn't exist.
exists = False
for app in apps:
if app["domain_name"] == domain:
exists = True
break
if not exists:
# Create the app.
response = requests.post(
path,
data={
"domain_name": domain,
"python_version": "python38",
},
headers=headers,
)
response.raise_for_status()
# Configure serving of static files.
path = self.url + "webapps/{domain}/static_files/".format(
domain=domain
)
response = requests.post(
path,
data={
"url": "/static/",
"path": self.static_path,
},
headers=headers,
)
response.raise_for_status()
counter += 1
# Upload files.
for target, source in self.files.items():
self.progress.setValue(counter)
with open(source, "rb") as source_file:
path = self.url + self.files_path + target
response = requests.post(
path,
files={"content": source_file.read()},
headers=headers,
)
response.raise_for_status()
counter += 1
# Update WSGI settings.
self.progress.setValue(counter)
logger.info(self.wsgi_config)
path = self.url + self.wsgi_path
response = requests.post(
path,
files={"content": self.wsgi_config},
headers=headers,
)
response.raise_for_status()
counter += 1
# Force HTTPS and source directory.
self.progress.setValue(counter)
data = {
"source_directory": "/home/{username}/{app_name}/".format(
username=self.username, app_name=self.app_name
),
"force_https": True,
}
path = self.url + "webapps/{domain}/".format(domain=domain)
response = requests.put(path, data=data, headers=headers)
response.raise_for_status()
counter += 1
# Reload the application.
self.progress.setValue(counter)
path = self.url + "webapps/{domain}/reload/".format(domain=domain)
response = requests.post(path, headers=headers)
response.raise_for_status()
counter += 1
self.progress.setValue(counter)
self.finished.emit(domain)
except Exception as ex:
# Gracefully report all failures (logged and reported upstream).
self.error.emit(repr(ex))