Commit a50344a5 authored by ale's avatar ale

import iniziale

tag_build = .dev
tag_svn_revision = true
from setuptools import setup, find_packages
description="Silly distributed URL shortener",
install_requires=["gevent", "greenlet", "bottle", "Jinja2", "formencode", "riak", "lrucache"],
package_data={"shorten":["templates/*", "static/*"]},
"console_scripts": [
import gevent
from gevent.pool import Pool
from gevent import monkey
import optparse
import urllib2
import urllib
import sys
import time
URL = ''
ENDPOINT = 'http://localhost:7000/'
status = {'ok': 0, 'error': 0}
def create_shortcut(n):
token = 'test%010d' % n
data = urllib.urlencode({'token': token, 'url': URL})
resp = urllib2.urlopen(ENDPOINT, data)
result =
status['ok'] += 1
except Exception, e:
print '%d: %s' % (n, str(e))
status['error'] += 1
def main():
parser = optparse.OptionParser()
parser.add_option('-c', '--concurrency', default='10', dest='concurrency',
help='Nr of concurrent connections to make.')
parser.add_option('-n', '--count', default='1000', dest='num_objs',
help='Nr of objects to create.')
opts, args = parser.parse_args()
if len(args) != 0:
parser.error('Wrong number of arguments.')
num_objs = int(opts.num_objs)
concurrency = int(opts.concurrency)
pool = Pool(concurrency)
start_time = time.time()
for n in xrange(num_objs):
pool.spawn(create_shortcut, n + 202910)
end_time = time.time()
elapsed = (end_time - start_time)
rate = num_objs / elapsed
print '%d objects, %d connections. ok=%d, err=%d' % (
num_objs, concurrency, status['ok'], status['error'])
print '%g elapsed, %.3g inserts/sec.' % (elapsed, rate)
if __name__ == '__main__':
"""Very simple anti-CSRF hack.
We generate a random token and sign it with HMAC, then set a cookie and
a hidden form field to the token and the signature respectively. If they
match at submission time, the request is validated.
The module includes a formencode-compatible validator to use in form
from shorten.wsgi import request, response
import formencode
from formencode import validators
import hmac
import base64
import hashlib
import os
COOKIE_NAME = '_shorten_frm'
def _digest(secret, data):
return hmac.HMAC(secret, data, hashlib.sha256).digest()
def gen_token(secret):
"""Generate a new token (and set the cookie)."""
a = os.urandom(8)
b = _digest(secret, a)
response.set_cookie(COOKIE_NAME, base64.b64encode(a))
return base64.b64encode(b)
def verify_token(secret, token):
"""Validate a token (against the cookie)."""
b = base64.b64decode(token)
a = base64.b64decode(request.COOKIES[COOKIE_NAME])
sig = _digest(secret, a)
return b == sig
class SecureFormValidator(validators.FancyValidator):
def __init__(self, secret):
self.secret = secret
def _to_python(self, value, state):
if not verify_token(self.secret, value):
raise formencode.Invalid('Invalid token', value, state)
return value
from shorten.wsgi import *
from shorten import secure_form
import riak.client
import formencode
from formencode import validators
import logging
import lrucache
import time
# Some configuration. Environment variables take precedence.
# You will most probably want to redefine RIAK_HOST and SECRET,
# at least.
RIAK_HOST = os.getenv('RIAK_HOST', '')
RIAK_BUCKET = os.getenv('RIAK_BUCKET', 'shorten')
CACHE_SIZE = int(os.getenv('CACHE_SIZE', 16384))
SECURE_FORM_SECRET = os.getenv('SECRET', 'flk2n30cn')
rdb = riak.client.RiakClient(host=RIAK_HOST)
url_bucket = rdb.bucket(RIAK_BUCKET)
cache = lrucache.LRUCache(CACHE_SIZE)
@route('/', method='GET')
def index():
return render('create_form.html',
class CreateSchema(formencode.Schema):
"""Validation schema for the new shortcut form."""
_frm = secure_form.SecureFormValidator(SECURE_FORM_SECRET)
url = validators.URL(add_http=True, check_exists=False, require_tld=True)
token = validators.Regex(r'^[a-zA-Z0-9]{3,16}$', not_empty=False)
@route('/', method='POST')
def create():
"""Create a new shortcut.
POST arguments:
token: (optional) the shortcut name, 3 to 16 alphanum chars
url: the destination url
form_data = CreateSchema.to_python(request.POST)
except formencode.Invalid, e:
return render('create_form.html', token=request.POST.get('token'),
url=request.POST.get('url'), error=str(e),
_frm=secure_form.gen_token(SECURE_FORM_SECRET))'create: %s' % (str(form_data),))
token = form_data['token']
url = form_data['url']
obj =, {'url': url, 'created_at': time.time()})
if obj.exists():'create: %s already exists' % (token,))
return render('create_form.html', url=url, error='Token already exists.',
_frm=secure_form.gen_token(SECURE_FORM_SECRET))'obj: %s' % (str(obj),))
# Store the object into Riak.
# Wait until 1 replica has confirmed the write. Discard the return body., dw=1, return_body=False)
return render('create_ok.html', token=token)
def resolve(token):
"""Resolve a shortcut.
If the token is found, redirect to the destination url.
Otherwise, present a form to the user to create it.
if token in cache:
url = cache[token]
obj = url_bucket.get(token)
if not obj.exists():
return render('create_form.html', token=token,
url = obj.get_data()['url']
cache[token] = url
redirect(url, 303)
if __name__ == '__main__':
body {
background: white;
color: #222;
font-family: sans-serif;
margin: 0;
padding: 1ex 4ex;
border-top: 6px solid #555;
h1 {
font-family: arial;
letter-spacing: -0.1em;
font-size: 36px;
} {
margin: 3px 20px 0 0;
border-bottom: 1px solid #333;
font-size: 10px;
table.summary {
font-size: 13px;
.report th {
color: #fff;
background: #333;
border-top: 1px solid #333;
border-bottom: 1px solid #333;
border-left: 1px solid #333;
padding: 1px 3px 1px 3px;
font-weight: normal;
text-align: left;
white-space: nowrap;
.report td {
border-top: 1px solid #d9d9d9;
border-left: 1px solid #d9d9d9;
padding: 3px;
vertical-align: top;
text-align: right;
white-space: nowrap;
.report td.last_row {
border-top: 3px solid #333;
font-weight: bold;
.report td.first_col {
border-left: none;
border-right: 1px solid #333;
font-weight: bold;
margin-left: 3px;
.report td.last_col {
font-weight: bold;
.report td.highlight {
background: #ff8;
#footer {
margin-top: 60px;
border-top: 1px solid #999;
text-align: right;
color: #999;
font-size: 80%;
<!DOCTYPE html
PUBLIC "-//W3C//DTD XHTML 1.0 Strict//EN"
<html xmlns="" lang="en" xml:lang="en">
<title>shorten :: {% block title %}{% endblock %}</title>
<link rel="shortcut icon" href="/static/favicon.ico" type="image/x-icon" />
<link rel="stylesheet" type="text/css" href="/static/style.css"/>
<h1>{{ self.title() }}</h1>
{% block main %}
{% endblock %}
<div id="footer">
shorten v{{ version }}@{{ hostname }}
<a href=""><img src="/static/"
width="80" height="15" alt="R*" /></a>
{% extends "_base.html" %}
{% block title %}
new shortcut
{% endblock %}
{% block main %}
{% if error %}
<p style="color:red;">{{ error | e }}</p>
{% endif %}
<form action="/" method="post">
Destination URL:<br/>
<input type="text" size="50" name="url" value="{{ url }}"/>
Shortcut (optional: if empty, it will be autogenerated):<br/>
<input type="text" size="16" name="token" value="{{ token }}"/>
<input type="submit" value="Create"/>
<input type="hidden" value="{{ _frm }}" name="_frm"/>
{% endblock %}
{% extends "_base.html" %}
{% block title %}
/{{ token }}
{% endblock %}
{% block main %}
The shortcut has been successfully created.<br/>
You can access it at
<a href="/{{ token }}">{{ token }}</a>
{% endblock %}
"""Minimal WSGI framework using Bottle and Jinja2."""
import gevent
from gevent import wsgi
from gevent import monkey
import collections
import logging
import os
import socket
from bottle import route, request, response, app, send_file, redirect, debug
from jinja2 import Environment, FileSystemLoader
__all__ = ['route', 'request', 'render', 'response', 'redirect', 'run_wsgi']
jinja = None
static_path = None
hostname = socket.gethostname()
version = '0.1'
def render(template, **args):
"""Render a template, with the provided context arguments.
template: template name (including the .html extension)
args: any keyword arguments will be available to the template
t = jinja.get_template(template)
targs = {'len': len, 'sorted': sorted, 'hostname': hostname, 'version': version}
return t.render(targs)
def serve_static_content(filename):
"""Method that serves static content out of 'static_path'."""
send_file(filename, root=static_path)
def run_wsgi(port, template_dir=None, static_dir=None):
"""Run the HTTP server.
port: TCP port to listen on.
template_dir: directory with templates (defaults to the 'templates'
subdirectory of the directory containing this file)
static_dir: directory with static content (mapped under /static,
defaults to the 'static' subdir)
global jinja, static_path
if not template_dir:
template_dir = os.path.join(
os.path.dirname(__file__), 'templates')
jinja = Environment(loader=FileSystemLoader(template_dir))
if not static_dir:
static_dir = os.path.join(os.path.dirname(__file__), 'static')
static_path = static_dir'starting http server on port %d' % port)
server = wsgi.WSGIServer(('', port), app())
server.backlog = 512
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment