Skip to content

Instantly share code, notes, and snippets.

@mushkevych

mushkevych/mbean_example.py

Created June 11, 2012 19:07
Show Gist options
  • Save mushkevych/2912007 to your computer and use it in GitHub Desktop.
Save mushkevych/2912007 to your computer and use it in GitHub Desktop.
Download ZIP
Python MX illustration example
Raw
mbean_example.py
class MBeanExample(object):
def __init__(self):
self.property_1 = 'this is'
self.property_2 = 'example'
def start_mx(self):
""" import MX module (which has back-reference import to self) and start it """
from mx.mx import MX
self.mx = MX(self)
self.mx.start_mx_thread()
if __name__ == '__main__':
source = MBeanExample()
source.start_mx()
Raw
mx.mx.py
from threading import Thread
from werkzeug.wrappers import Request
from werkzeug.wsgi import ClosingIterator, SharedDataMiddleware
from werkzeug.exceptions import HTTPException, NotFound
from werkzeug.serving import run_simple
from utils import STATIC_PATH, local, local_manager, url_map, jinja_env
import views
class MX(object):
def __init__(self, mbean):
local.application = self
self.mbean = mbean
jinja_env.globals['mbean'] = mbean
self.dispatch = SharedDataMiddleware(self.dispatch, {
'/static': STATIC_PATH
})
def dispatch(self, environ, start_response):
local.application = self
request = Request(environ)
local.url_adapter = adapter = url_map.bind_to_environ(environ)
try:
endpoint, values = adapter.match()
handler = getattr(views, endpoint)
response = handler(request, **values)
except NotFound as e:
response = views.not_found(request)
response.status_code = 404
except HTTPException as e:
response = e
return ClosingIterator(response(environ, start_response),
[local_manager.cleanup])
def __call__(self, environ, start_response):
return self.dispatch(environ, start_response)
def start_mx_thread(self, hostname = None, port = None):
"""Spawns a new HTTP server, residing on defined hostname and port
:param hostname: the default hostname the server should listen on.
:param port: the default port of the server.
"""
hostname = '0.0.0.0'
port = 5000
reloader = False # use_reloader: the default setting for the reloader.
debugger= False #
evalex=True # use_evalex: the default setting for the evalex flag of the debugger.
threaded=False # threaded: the default threading setting.
processes=1 # processes: the default number of processes to start.
reloader_interval = 1
static_files=None # static_files: optional dict of static files.
extra_files=None # extra_files: optional list of extra files to track for reloading.
ssl_context=None # ssl_context: optional SSL context for running server in HTTPS mode.
mx_thread = Thread(target=run_simple(hostname = hostname,
port = port,
application = self,
use_debugger = debugger,
use_evalex = evalex,
extra_files = extra_files,
use_reloader = reloader,
reloader_interval = reloader_interval,
threaded = threaded,
processes = processes,
static_files=static_files,
ssl_context=ssl_context))
mx_thread.daemon = True
mx_thread.start()
Raw
mx.templates.not_found.html
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN"
"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<body>
<h2>Page Not Found</h2>
<p>
The page you have requested does not exist on this server.
</p>
</body>
</html>
Raw
mx.templates.template_example.html
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN"
"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<body>
<table style="width: 100%"> <tbody>
{%- for row in details.entries %}
<tr>
<td>{{ row[0] }}</td>
<td>{{ row[1] }}</td>
</tr>
{%- endfor %}
</tbody> </table>
</body>
</html>
Raw
mx.utils.py
from datetime import datetime
from os import path
from urlparse import urlparse
from jinja2 import Environment, FileSystemLoader
from werkzeug.local import Local, LocalManager
from werkzeug.wrappers import Response
from werkzeug.routing import Map, Rule
TEMPLATE_PATH = path.join(path.dirname(__file__), 'templates')
STATIC_PATH = path.join(path.dirname(__file__), 'static')
ALLOWED_SCHEMES = frozenset(['http', 'https', 'ftp', 'ftps'])
URL_CHARS = 'abcdefghijkmpqrstuvwxyzABCDEFGHIJKLMNPQRST23456789'
local = Local()
local_manager = LocalManager([local])
url_map = Map([Rule('/static/<file>', endpoint='static', build_only=True)])
def expose(rule, **kw):
def decorate(f):
kw['endpoint'] = f.__name__
url_map.add(Rule(rule, **kw))
return f
return decorate
def url_for(endpoint, _external=False, **values):
return local.url_adapter.build(endpoint, values, force_external=_external)
def render_template(template, **context):
return Response(jinja_env.get_template(template).render(**context),
mimetype='text/html')
def validate_url(url):
return urlparse(url)[0] in ALLOWED_SCHEMES
jinja_env = Environment(loader=FileSystemLoader(TEMPLATE_PATH))
jinja_env.globals['url_for'] = url_for
jinja_env.globals['local'] = local
Raw
mx.views.py
from werkzeug.utils import cached_property, redirect
from werkzeug.wrappers import Response
from utils import render_template, expose, jinja_env
@expose('/')
@expose('/example_details/')
def example_details(request):
details = ExampleDetails(jinja_env.globals['mbean'])
return render_template('template_example.html', details=details)
def not_found(request):
return render_template('not_found.html')
# And implement details themselves:
class ExampleDetails(object):
def __init__(self, mbean):
self.mbean = mbean
@cached_property
def entries(self):
return [[self.mbean.property_1, self.mbean.property_2]]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment