WEBHOOKS README
Overview
This document explains my thought process, design considerations, and
implementation details for the webhook system task.
Understanding Webhooks
Before diving into the task, I took a moment to refresh my understanding of what
webhooks are.
A webhook is a way for one system to send real-time data or notifications to another
system whenever a specific event occurs. Think of it as a one-way communication
method triggered by an event. Unlike a traditional API, which requires requests to
fetch data, webhooks are event-driven and push data to a specified endpoint.
System Requirements
Functional Requirements
Here’s what the system needs to do:
Allow users to register and deregister webhooks.
Notify registered users when an event they subscribed to is triggered.
Non-Functional Requirements
I also considered some non-functional requirements to ensure the system works
reliably:
High Availability: The system should be accessible and operational at all
times.
Scalability: It should handle a growing number of clients and events.
Security: Validations should ensure only legitimate requests are processed.
Error Handling: The system should log errors and provide fallback
mechanisms for failed broadcasts.
High-Level Design
Design Considerations
I approached the system design with simplicity and maintainability in mind while
ensuring it could handle future enhancements. Here's what guided my decisions:
1. The system functions similarly to a notification service, so I designed it with
that in mind.
2. A rate limiter could be useful to manage the number of messages sent to
client servers, especially if we implement retries for failed requests later.
High Level Design Consideration
[Link]
Workflow Overview
Registration Flow:
Client → API → Authentication → Validation → Register/Deregister
Webhooks
Broadcast Flow:
Event Trigger → Broadcast Controller → Request Handler → Target Client
Endpoints
Error Handling Flow:
Worker → Error Handler → Error Logs
Testing Approach
Testing was an important part of the process to ensure the system works as expected.
Here's how I approached it:
Manual Testing:
I tested the endpoints using Thunder Client to confirm they behaved as expected.
Unit Tests:
Verified webhook registration and deregistration logic.
Checked URL formatting and event type validation.
Integration Tests:
Tested the full flow of broadcasting events to registered web-hooks.
Simulated error scenarios to ensure the system logs errors and gracefully handles
failures.
Possible System Enhancements
Given additional time, the system could be enhanced with:
Message queue integration for reliable delivery
Enhanced security features
Retry Mechanism.
MY CODE IMPLEMENTATION FROM LINE 197
class User([Link]):
"""
In other to show the implementation and how paramount security is i created a user
model,
leftout the relationship model with the webhook so it doesnt get confusing
"""
__tablename__ = 'users'
is_active = [Link]([Link], default=True)
id = [Link]([Link], primary_key=True)
email = [Link]([Link])
password = [Link]([Link](200), nullable=False)
secret = [Link]([Link])
def __repr__(self):
return [Link]
class Webhooks([Link]):
__tablename__ = 'webhooks'
id = [Link]([Link], primary_key=True)
url = [Link]([Link])
ntype = [Link]([Link])
def __init__(self, url, ntype):
[Link] = url
[Link] = ntype
def __repr__(self):
return f"{[Link]} with {[Link]} Registered"
#Making use of Pydantic for validation to ensure data integrity
class UserSignup(BaseModel):
email: EmailStr # Automatically validates email format
password: str = Field(..., min_length=8, max_length=100)
##
@base_blueprint.route('/api/signup', methods=['POST'])
def signup():
"""
Handle user registration endpoint.
The data is validated using Pydantic model UserSignup.
Passwords are securely hashed using bcrypt before storage.
A unique secret key is generated for each user.
I implemented this to help out with security of the application
generating a unique key for every user, so we can identify users making request to
our server
"""
try:
data = request.get_json()
validated_data = UserSignup(**data) # Validate the data using the Pydantic
model
email = validated_data.email
password = validated_data.password
secret_key = secrets.token_hex(32)
hashed_password = bcrypt.generate_password_hash(password).decode('utf-8')
new_user = User(
email=validated_data.email,
password=hashed_password,
secret=secret_key
)
[Link](new_user)
[Link]()
return jsonify({
"message": "Signup successful!",
"email": email }), 200
except ValidationError as e:
# Return validation errors as JSON
return jsonify({
"error": "Invalid input",
"details": [Link]() # List of validation errors
}), 400
@base_blueprint.route('/login', methods=['POST'])
def login():
try:
email = [Link]['email']
password = [Link]['password']
except Exception as e:
return jsonify({
"error": "Invalid input",
"details": str(e) # List of validation errors
}), 400
try:
user = [Link].filter_by(email=email).first()
hashed_password = [Link]
#bcrypt is used for password hashing
bcrypt.check_password_hash(hashed_password, password)
access_token = create_access_token(identity=[Link])
return {'access_token': access_token}, 200
except Exception as e:
return jsonify({
"error": "Invalid input",
"details": str(e) # List of validation errors
}), 400
def validate_webhook_url(url: str) -> Optional[str]:
"""
A webhook helper function that helps validate url
"""
try:
result = urlparse(url)
if all([[Link] in ['http', 'https'], [Link]]):
return "Valid"
return "Invalid URL scheme. Only HTTP(S) URLs are allowed."
except Exception:
return "Invalid URL format"
# Webhook registration endpoint
@base_blueprint.route('/webhook/register', methods=['POST'])
@jwt_required()
def register_webhook() -> Response:
"""
Register a new webhook endpoint for notifications.
Requires JWT authentication. Validates the user's secret key,
webhook URL format, and notification type before registration.
"""
data = [Link]
email = get_jwt_identity()
user_secret = [Link].filter_by(email=email).first().secret
url = data['url']
received_secret = data['secret']
# Validate required fields
for params in data:
if params not in ['url', 'ntype', 'secret']:
raise InvalidUsage('Missing required fields', status_code=400)
validate_webhook_url(url)
if user_secret != received_secret:
raise InvalidUsage(f'Invalid secret key. You are not authorised to register a
webhook',
status_code=400)
if data['ntype'] not in notification_types:
raise InvalidUsage(f'Invalid event type. Must be one of: {notification_types}',
status_code=400)
try:
url = data['url']
ntype = data['ntype']
add_db_webhook(
url=data['url'],
ntype=data['ntype']
)
return jsonify({
'status': 'success',
'message': 'Webhook registered successfully'
}), 201
except Exception as e:
raise InvalidUsage(f'Failed to register webhook: {str(e)}', status_code=500)
@base_blueprint.route('/api/event', methods=['GET'])
def trigger_event():
"""
Test endpoint to simulate event broadcasting.
Triggers webhook notifications based on provided event type.
"""
data = request.get_json()
ntype = [Link]('ntype')
broadcast_webhook(ntype=ntype)
return jsonify({
'status': 'success',
'message': 'Webhook Triggered'
}), 200
@base_blueprint.route('/webhook/deregister', methods=['DELETE'])
@jwt_required()
def deregister_webhook() -> Response:
"""
Deregister an existing webhook endpoint.
Requires JWT authentication and validates the user's secret key
before removing the webhook registration.
"""
data = request.get_json()
url = data['url']
ntype = data['ntype']
for params in data:
if params not in ['url', 'ntype', 'secret']:
raise InvalidUsage('Missing required fields', status_code=400)
email = get_jwt_identity()
user_secret = [Link].filter_by(email=email).first().secret
received_secret = data['secret']
if user_secret != received_secret:
raise InvalidUsage(f'Invalid secret key. You are not authorised to register a
webhook',
status_code=400)
if data['ntype'] not in notification_types:
raise InvalidUsage(f'Invalid event type. Must be one of: {notification_types}',
status_code=400)
try:
remove_db_webhook(url, ntype)
return jsonify({
'status': 'success',
'message': 'Webhook deregistered successfully'
}), 200
except Exception as e:
raise InvalidUsage(f'Failed to deregister webhook: {str(e)}', status_code=500)
def webhook_exception_handler(request, exception):
"""Handle any exceptions that occur during the request"""
return f"Request to {[Link]} failed: {str(exception)}"
@[Link]("10 per minute")
def broadcast_webhook(ntype: str) -> None:
"""
Broadcasts events to all registered webhooks of specified type.
Uses grequests for efficient parallel HTTP requests to webhooks.
Handles timeouts and failures gracefully for each endpoint.
Rate limited to 10 broadcasts per minute to prevent system overload.
Args:
ntype: Type of notification/event to broadcast
Implementation details:
- Makes parallel requests using grequests with 5 concurrent connections
- Sets 10-second timeout for each webhook call
- Tracks success/failure for each endpoint
- Returns detailed response info including status codes
Note: Failed requests are logged but don't stop other webhooks from
receiving the event. This ensures one bad endpoint doesn't break
notifications for everyone else.
"""
webhooks = get_ntype_webhooks(ntype)
requests = []
for webhook in webhooks:
# Create request with headers and timeout
try:
req = [Link](
url=[Link],
json={
'event_time':
[Link]([Link]).strftime("%d/%b/%Y:%H:%M:%S.%f")[:-7],
},
headers={'Content-Type': 'application/json'},
timeout=10
)
[Link](req)
responses = [Link](
requests,
exception_handler=webhook_exception_handler,
size=5 # Number of simultaneous requests
)
except Exception as e:
print(e)
return f"Error {e}"
results = []
for webhook, response in zip(webhooks, responses):
if type(response) != str:
[Link]({
'url': [Link],
'success': 200 <= response.status_code < 300,
'status_code': response.status_code,
'response': [Link]
})
else:
[Link]({
'url': [Link],
'success': False,
'status_code': None ,
'error': 'Request failed'
})
@base_blueprint.route('/webhook/receive', methods=['POST'])
def receive_webhook():
"""
Test endpoint that simulates a client server receiving webhook events.
Useful for testing the webhook delivery system.
The endpoint logs received data and returns it in the response
to verify successful delivery.
Note: This is for testing/debugging only and simulates how a real
client endpoint would receive and handle webhook events.
"""
try:
# Get the webhook payload
data = request.get_json()
# Log the received webhook
[Link](f"Received webhook: {data}")
# Process webhook data
event_time = [Link]('event_time')
payload = [Link]('payload')
# Return success response
return jsonify({
'status': 'success',
'message': 'Webhook received and processed successfully',
'received_data': {
'event_time': event_time,
'payload': payload
}
}), 200
except Exception as e:
[Link](f"Error processing webhook: {str(e)}")
return jsonify({
'status': 'error',
'message': f'Failed to process webhook: {str(e)}'
}), 500
if __name__ == '__main__':
create_app().run(debug=True)
[Link]
alembic==1.14.0
annotated-types==0.7.0
bcrypt==4.2.1
blinker==1.9.0
certifi==2024.8.30
cffi==1.17.1
charset-normalizer==3.4.0
click==8.1.7
colorama==0.4.6
decorator==5.1.1
Deprecated==1.2.15
dnspython==2.7.0
email_validator==2.2.0
Flask==3.1.0
Flask-BasicAuth==0.2.0
Flask-Bcrypt==1.0.1
Flask-Cors==5.0.0
Flask-JWT-Extended==4.7.1
Flask-Limiter==3.9.2
Flask-Login==0.6.3
Flask-Migrate==4.0.7
Flask-SQLAlchemy==3.1.1
gevent==24.11.1
greenlet==3.1.1
grequests==0.7.0
idna==3.10
itsdangerous==2.2.0
Jinja2==3.1.4
limits==3.14.1
Mako==1.3.7
markdown-it-py==3.0.0
MarkupSafe==3.0.2
mdurl==0.1.2
ordered-set==4.1.0
packaging==24.2
pycparser==2.22
pydantic==2.10.3
pydantic_core==2.27.1
Pygments==2.18.0
PyJWT==2.10.1
python-dotenv==1.0.1
requests==2.32.3
rich==13.9.4
setuptools==75.6.0
SQLAlchemy==2.0.36
typing_extensions==4.12.2
urllib3==2.2.3
Werkzeug==3.1.3
wrapt==1.17.0
[Link]==5.0
[Link]==7.2