Browse Source

remove extra spaces before newlines (no code/logic changes)

test-message
Trent Larson 9 months ago
parent
commit
140e426458
  1. 34
      app.py

34
app.py

@ -27,15 +27,15 @@ class WebPushService():
""" """
This class provides services for sending web push notifications. This class provides services for sending web push notifications.
""" """
def __init__(self, app, config_name: str) -> None: def __init__(self, app, config_name: str) -> None:
""" """
Initializes the WebPushService with the given application and configuration name. Initializes the WebPushService with the given application and configuration name.
Args: Args:
- app: The application instance where the service will be attached. - app: The application instance where the service will be attached.
- config_name (str): The name of the configuration to be used. - config_name (str): The name of the configuration to be used.
Attributes: Attributes:
- app: The application instance. - app: The application instance.
- daily_notification_thread (threading.Thread): A thread to send daily notifications. - daily_notification_thread (threading.Thread): A thread to send daily notifications.
@ -76,7 +76,7 @@ class WebPushService():
Raises: Raises:
- Exceptions raised by the key generation or database operations are caught and printed. - Exceptions raised by the key generation or database operations are caught and printed.
""" """
try: try:
# Generating a private key using the elliptic curve SECP256R1 # Generating a private key using the elliptic curve SECP256R1
private_key = ec.generate_private_key(ec.SECP256R1(), default_backend()) private_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
@ -104,7 +104,7 @@ class WebPushService():
except Exception as e: except Exception as e:
print(f"Error generating VAPID keys: {e}") print(f"Error generating VAPID keys: {e}")
def _initialize(self) -> None: def _initialize(self) -> None:
""" """
Initializes the WebPushService by checking for the presence of VAPID keys in the database. Initializes the WebPushService by checking for the presence of VAPID keys in the database.
@ -119,7 +119,7 @@ class WebPushService():
# Checking if there are any VAPID keys in the database # Checking if there are any VAPID keys in the database
if not VAPIDKey.query.first(): if not VAPIDKey.query.first():
# Generating and saving VAPID keys if none are found # Generating and saving VAPID keys if none are found
self._generate_and_save_vapid_keys() self._generate_and_save_vapid_keys()
@ -152,7 +152,7 @@ class WebPushService():
) )
time.sleep(1) time.sleep(1)
return True return True
except WebPushException as ex: except WebPushException as ex:
now = datetime.datetime.now().isoformat() now = datetime.datetime.now().isoformat()
endpoint = subscription_info['endpoint'] endpoint = subscription_info['endpoint']
@ -191,15 +191,15 @@ class WebPushService():
- A context for the application is created to enable database operations. - A context for the application is created to enable database operations.
- The message content for the daily update is hardcoded in this method. - The message content for the daily update is hardcoded in this method.
""" """
while True: while True:
now = datetime.datetime.now().isoformat() now = datetime.datetime.now().isoformat()
print(f"{now} - Starting to send subscriptions...", flush=True) print(f"{now} - Starting to send subscriptions...", flush=True)
# Creating a context for the application to enable database operations # Creating a context for the application to enable database operations
with self.app.app_context(): with self.app.app_context():
# Retrieving all subscription data from the database # Retrieving all subscription data from the database
all_subscriptions = Subscription.query.all() all_subscriptions = Subscription.query.all()
@ -245,7 +245,7 @@ class WebPushService():
- If the operation is successful, a JSON response with a success message is returned with a 200 status code. - If the operation is successful, a JSON response with a success message is returned with a 200 status code.
- If there's an error during the operation, a JSON response with the error message is returned with a 500 status code. - If there's an error during the operation, a JSON response with the error message is returned with a 500 status code.
""" """
# This default can be invoked thus: curl -X POST -H "Authorization: Basic YWRtaW46YWRtaW4=" localhost:3000/web-push/regenerate_vapid # This default can be invoked thus: curl -X POST -H "Authorization: Basic YWRtaW46YWRtaW4=" localhost:3000/web-push/regenerate_vapid
envPassword = os.getenv('ADMIN_PASSWORD', 'admin') envPassword = os.getenv('ADMIN_PASSWORD', 'admin')
auth = request.authorization auth = request.authorization
@ -269,9 +269,9 @@ class WebPushService():
# Generating and saving new VAPID keys # Generating and saving new VAPID keys
self._generate_and_save_vapid_keys() self._generate_and_save_vapid_keys()
return jsonify(success=True, message="VAPID keys regenerated successfully"), 200 return jsonify(success=True, message="VAPID keys regenerated successfully"), 200
except Exception as e: except Exception as e:
return jsonify(error=f'Error regenerating VAPID keys: {str(e)}'), 500 return jsonify(error=f'Error regenerating VAPID keys: {str(e)}'), 500
@ -296,7 +296,7 @@ class WebPushService():
# Retrieving the VAPID key from the database # Retrieving the VAPID key from the database
key = VAPIDKey.query.first() key = VAPIDKey.query.first()
# Returning the public key in a JSON response # Returning the public key in a JSON response
return jsonify(vapidKey=key.public_key) return jsonify(vapidKey=key.public_key)
@ -359,14 +359,14 @@ class WebPushService():
# Creating a confirmation message for the push notification # Creating a confirmation message for the push notification
message = {"title": "Subscription Successful", "message": "Thank you for subscribing!"} message = {"title": "Subscription Successful", "message": "Thank you for subscribing!"}
# Sending the confirmation push notification # Sending the confirmation push notification
success = WebPushService._send_push_notification(subscription_info, message, vapid_key) success = WebPushService._send_push_notification(subscription_info, message, vapid_key)
# Returning the operation status # Returning the operation status
return jsonify(success=success) return jsonify(success=success)
@staticmethod @staticmethod
@app.route('/web-push/unsubscribe', methods=['POST']) @app.route('/web-push/unsubscribe', methods=['POST'])
def unsubscribe() -> Tuple[str, int]: def unsubscribe() -> Tuple[str, int]:
@ -401,7 +401,7 @@ class WebPushService():
db.session.delete(subscription) db.session.delete(subscription)
db.session.commit() db.session.commit()
return jsonify(success=True, message="Subscription deleted successfully") return jsonify(success=True, message="Subscription deleted successfully")
# If the subscription is not found, return an error message # If the subscription is not found, return an error message
else: else:
return jsonify(success=False, error="Subscription not found"), 404 return jsonify(success=False, error="Subscription not found"), 404

Loading…
Cancel
Save