diff --git a/.gitignore b/.gitignore index b25c15b..479ce02 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ *~ +.aider* diff --git a/README.md b/README.md index 647f375..84e3747 100644 --- a/README.md +++ b/README.md @@ -32,7 +32,7 @@ sudo docker run -d -p 8900:3000 -v ~/py-push-server-db:/app/instance/data --name On a production server for security (eg /web-push/generate_vapid): set an environment variable `ADMIN_PASSWORD` for permissions; one way is to add this to the `docker run` command: `-e ADMIN_PASSWORD=` -Finally, after it's started, generate a new VAPID by hitting the `regenerate_vapid` endpoint with a POST, eg. `curl -X POST localhost:8080/web-push/regenerate_vapid` +Finally, after it's started, generate a new VAPID by hitting the `regenerate-vapid` endpoint with a POST, eg. `curl -X POST localhost:8080/web-push/regenerate-vapid` @@ -244,7 +244,10 @@ pip install gunicorn source venv/bin/activate -gunicorn -b 0.0.0.0:3000 --log-level=debug --workers=1 app:app # 3 workers trigger 3 daily subscription runs +cp data/webpush.db.empty data/webpush.db + +# 3 workers would trigger 3 daily subscription runs +gunicorn -b 0.0.0.0:3000 --log-level=debug --workers=1 app:app ``` diff --git a/app.py b/app.py index 2c0e8ee..815e0a8 100644 --- a/app.py +++ b/app.py @@ -43,7 +43,7 @@ class WebPushService(): # Setting the application instance self.app = app - self.app.add_url_rule('/web-push/regenerate_vapid', view_func=self.regenerate_vapid, methods=['POST']) + self.app.add_url_rule('/web-push/regenerate-vapid', view_func=self.regenerate_vapid, methods=['POST']) # Setting the database URI for the application db_uri = os.getenv('SQLALCHEMY_DATABASE_URI', 'sqlite:////app/instance/data/webpush.db') @@ -102,7 +102,8 @@ class WebPushService(): db.session.commit() except Exception as e: - print(f"Error generating VAPID keys: {e}") + print(f"Error generating VAPID keys: {str(e)}") + raise e def _initialize(self) -> None: @@ -135,7 +136,8 @@ class WebPushService(): - vapid_key (VAPIDKey): The VAPID key model instance containing the private key used for sending the notification. Returns: - - bool: True if the push notification was sent successfully, False otherwise. + - request.Response https://requests.readthedocs.io/en/latest/api.html#requests.Response if the push notification was sent successfully + or False if there was an exception Notes: - The `webpush` function is used to send the notification. @@ -144,14 +146,15 @@ class WebPushService(): # Sending the push notification using the webpush function try: - webpush( + result = webpush( subscription_info=subscription_info, data=json.dumps(message), vapid_private_key=vapid_key.private_key, vapid_claims={"sub": CONTACT_EMAIL} ) + # "because sometimes that's what I had to do to make it work!" - Matthew time.sleep(1) - return True + return {"success": True, "message": result.text, "result": result} except WebPushException as ex: now = datetime.datetime.now().isoformat() @@ -172,7 +175,7 @@ class WebPushService(): else: print("Error other than unsubscribed/expired.", ex.args[0], flush=True) - return False + return {"success": False, "message": ex.args[0]} def _send_daily_notifications(self) -> None: @@ -234,19 +237,19 @@ class WebPushService(): 1. Deletes the current VAPID keys from the database. 2. Generates and stores new VAPID keys using the `_generate_and_save_vapid_keys` method. - URL: /web-push/regenerate_vapid + URL: /web-push/regenerate-vapid Method: POST Header: Authentication: Basic ... Returns: - - Tuple[str, int]: A JSON response indicating the success or failure of the operation, along with the appropriate HTTP status code. + - tuple with "success" as True or False, and "message" message string Notes: - 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. """ - # 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') auth = request.authorization if (auth is None @@ -273,7 +276,7 @@ class WebPushService(): return jsonify(success=True, message="VAPID keys regenerated successfully"), 200 except Exception as e: - return jsonify(error=f'Error regenerating VAPID keys: {str(e)}'), 500 + return jsonify(success=False, message=f'Error regenerating VAPID keys: {str(e)}'), 500 @staticmethod @@ -317,7 +320,7 @@ class WebPushService(): Method: POST Returns: - - Tuple[str, int]: A JSON response indicating the success or failure of the operation, along with the appropriate HTTP status code. + - A JSON response with "success" as True or False, "message" as a response message string, and potentially a "result" as a request.Response object from sending a test notification Notes: - If the operation is successful, a confirmation push notification is sent to the subscriber with a success message. @@ -333,7 +336,7 @@ class WebPushService(): # Checking if the VAPID key is available if not vapid_key: - return jsonify(success=False, error="No VAPID keys available"), 500 + return jsonify(success=False, message="No VAPID keys available"), 500 # Creating a new Subscription instance with the provided data subscription = Subscription(endpoint=content['endpoint'], @@ -361,10 +364,10 @@ class WebPushService(): message = {"title": "Subscription Successful", "message": "Thank you for subscribing!"} # Sending the confirmation push notification - success = WebPushService._send_push_notification(subscription_info, message, vapid_key) + result = WebPushService._send_push_notification(subscription_info, message, vapid_key) # Returning the operation status - return jsonify(success=success) + return jsonify(success=result["status_code"]==201, message=result["text"]) @staticmethod @@ -404,7 +407,69 @@ class WebPushService(): # If the subscription is not found, return an error message else: - return jsonify(success=False, error="Subscription not found"), 404 + return jsonify(success=False, message="Subscription not found"), 404 + + + @staticmethod + @app.route('/web-push/send-test', methods=['POST']) + def send_test() -> Tuple[str, int]: + """ + Endpoint to send a test push notification to a specific client. + + This method: + 1. Retrieves the subscription information from the incoming request. + 2. Looks up the subscription in the database. + 3. Calls the _send_push_notification method to send a test push notification. + 4. Returns the result of the _send_push_notification call. + + URL: /web-push/send-test + Method: POST + + Returns: + - A JSON response with the result of the _send_push_notification call. + + Notes: + - The incoming request should contain the parameters "endpoint", "p256dh", and "auth". + """ + + # Retrieving the subscription information from the incoming request + content = request.json + endpoint = content['endpoint'] + p256dh = content['keys']['p256dh'] + auth = content['keys']['auth'] + + # Looking up the subscription in the database + subscription = Subscription.query.filter_by(endpoint=endpoint, p256dh=p256dh, auth=auth).first() + + # If the subscription is found, call the _send_push_notification method + if subscription: + subscription_info = { + "endpoint": subscription.endpoint, + "keys": { + "p256dh": subscription.p256dh, + "auth": subscription.auth + } + } + + title = "Test Notification" + if "title" in content: + title = content['title'] + message = "This is a test notification." + if "message" in content: + message = content['message'] + + vapid_key = VAPIDKey.query.filter_by(id=subscription.vapid_key_id).first() + result = WebPushService._send_push_notification( + subscription_info, + {"title": title, "message": message}, + vapid_key + ) + return jsonify( + success=result["result"].status_code==201, + message=result["result"].text, + ) + else: + return jsonify({"success": False, "message": "Subscription not found"}), 404 web_push_service = WebPushService(app, "app")