"""
Demo resource for the OpenEventDatabase.
This module imports and re-exports the demo resources from the demo package.
"""
import falcon
import requests
import json
import os
from collections import defaultdict
from oedb.utils.logging import logger
from oedb.utils.db import load_env_from_file
from oedb.resources.demo import demo_main, demo_traffic, demo_view_events
class DemoResource:
"""
Resource for the demo endpoint.
Handles the /demo endpoint and related demo pages.
"""
def on_get_edit(self, req, resp, id=None):
"""
Handle GET requests to the /demo/edit endpoint.
Returns an HTML page with a form for editing an existing event.
Args:
req: The request object.
resp: The response object.
id: The event ID to edit.
"""
logger.info(f"Processing GET request to /demo/edit for event ID: {id}")
if id is None:
resp.status = falcon.HTTP_400
resp.text = "Event ID is required"
return
try:
# Set content type to HTML
resp.content_type = 'text/html'
# Fetch the event data from the API
response = requests.get(f'https://api.openeventdatabase.org/event/{id}')
if response.status_code != 200:
resp.status = falcon.HTTP_404
resp.text = f"Event with ID {id} not found"
return
event_data = response.json()
# Create HTML response with form
html = f"""
"""
# Set the response body and status
resp.text = html.replace('{event_data}', json.dumps(event_data))
resp.status = falcon.HTTP_200
logger.success(f"Successfully processed GET request to /demo/edit for event ID: {id}")
except Exception as e:
logger.error(f"Error processing GET request to /demo/edit: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
def on_get(self, req, resp):
"""
Handle GET requests to the /demo endpoint.
Delegates to the demo_main resource.
Args:
req: The request object.
resp: The response object.
"""
return demo_main.on_get(req, resp)
def on_get_by_what(self, req, resp):
"""
Handle GET requests to the /demo/by-what endpoint.
Returns an HTML page with links to events organized by their "what" type.
Args:
req: The request object.
resp: The response object.
"""
logger.info("Processing GET request to /demo/by-what")
try:
# Set content type to HTML
resp.content_type = 'text/html'
# Fetch events from the API
try:
response = requests.get('/event?limit=1000')
if response.status_code == 200 and response.text:
events_data = response.json()
else:
logger.error(f"Error fetching events: Status code {response.status_code}, Response: {response.text}")
events_data = {"features": []}
except json.JSONDecodeError as e:
logger.error(f"Error parsing JSON response: {e}")
events_data = {"features": []}
except Exception as e:
logger.error(f"Error fetching events: {e}")
events_data = {"features": []}
# Group events by "what" type
events_by_what = defaultdict(list)
if events_data.get('features'):
for feature in events_data['features']:
properties = feature.get('properties', {})
what = properties.get('what', 'Unknown')
events_by_what[what].append({
'id': properties.get('id'),
'label': properties.get('label', 'Unnamed Event'),
'coordinates': feature.get('geometry', {}).get('coordinates', [0, 0])
})
# Create HTML response
html = """
Events by Type - OpenEventDatabase
"
html += """
"""
# Set the response body and status
resp.text = html
resp.status = falcon.HTTP_200
logger.success("Successfully processed GET request to /demo/by-what")
except Exception as e:
logger.error(f"Error processing GET request to /demo/by-what: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
def on_get_search(self, req, resp):
"""
Handle GET requests to the /demo/search endpoint.
Returns an HTML page with a form for searching events and displaying results.
Args:
req: The request object.
resp: The response object.
"""
logger.info("Processing GET request to /demo/search")
try:
# Set content type to HTML
resp.content_type = 'text/html'
# Create HTML response with search form
html = """
Search Events - OpenEventDatabase
"""
# Set the response body and status
resp.text = html
resp.status = falcon.HTTP_200
logger.success("Successfully processed GET request to /demo/search")
except Exception as e:
logger.error(f"Error processing GET request to /demo/search: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
def on_get_map_by_what(self, req, resp):
"""
Handle GET requests to the /demo/map-by-what endpoint.
Returns an HTML page with a MapLibre map showing events filtered by "what" type.
Args:
req: The request object.
resp: The response object.
"""
logger.info("Processing GET request to /demo/map-by-what")
try:
# Set content type to HTML
resp.content_type = 'text/html'
# Create HTML response with MapLibre map and filtering controls
html = """
Map by Event Type - OpenEventDatabase
This map shows events from the OpenEventDatabase filtered by their type.
Use the filter panel on the right to show/hide different event types.
Loading events...
Filter by Event Type
Loading event types...
"""
# Set the response body and status
resp.text = html
resp.status = falcon.HTTP_200
logger.success("Successfully processed GET request to /demo/map-by-what")
except Exception as e:
logger.error(f"Error processing GET request to /demo/map-by-what: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
events_by_what = defaultdict(list)
if events_data.get('features'):
for feature in events_data['features']:
properties = feature.get('properties', {})
what = properties.get('what', 'Unknown')
events_by_what[what].append({
'id': properties.get('id'),
'label': properties.get('label', 'Unnamed Event'),
'coordinates': feature.get('geometry', {}).get('coordinates', [0, 0])
})
# Create HTML response
html = """
Events by Type - OpenEventDatabase
"
html += """
"""
# Set the response body and status
resp.text = html
resp.status = falcon.HTTP_200
logger.success("Successfully processed GET request to /demo/by-what")
except Exception as e:
logger.error(f"Error processing GET request to /demo/by-what: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
def on_get_traffic(self, req, resp):
"""
Handle GET requests to the /demo/traffic endpoint.
Delegates to the demo_traffic resource.
Args:
req: The request object.
resp: The response object.
"""
return demo_traffic.on_get(req, resp)
def on_get_view_events(self, req, resp):
"""
Handle GET requests to the /demo/view-events endpoint.
Delegates to the demo_view_events resource.
Args:
req: The request object.
resp: The response object.
"""
return demo_view_events.on_get(req, resp)
def on_get_by_id(self, req, resp, id):
"""
Handle GET requests to /demo/by_id/{id}.
Show a map with the event location and a table of its properties.
"""
import requests
logger.info(f"Processing GET request to /demo/by_id/{id}")
try:
resp.content_type = 'text/html'
r = requests.get(f"https://api.openeventdatabase.org/event/{id}")
r.raise_for_status()
feature = r.json()
html = f"""
Event {id} - OpenEventDatabase
' for k,v in sorted((feature.get('properties') or {{}}).items())])}
"""
resp.text = html
resp.status = falcon.HTTP_200
logger.success(f"Successfully processed GET request to /demo/by_id/{id}")
except Exception as e:
logger.error(f"Error processing GET request to /demo/by_id/{id}: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
# Create a global instance of DemoResource
demo = DemoResource()