Fun stuff I wrote with flask-peewee

bookmarks.py

"""
Simple bookmarks page that's bookmarklet-friendly
"""

import datetime

from flask import request, redirect, url_for, render_template, Blueprint
from peewee import *

from flaskext.rest import RestResource
from flaskext.utils import get_object_or_404, object_list

from api import api
from app import app, db
from auth import auth


class Bookmark(db.Model):
    url = TextField()
    created_date = DateTimeField(default=datetime.datetime.now)

    class Meta:
        ordering = (('created_date', 'desc'),)

    def __unicode__(self):
        return self.url

bookmarks = Blueprint('bookmarks', __name__, template_folder='templates')

@bookmarks.route('/')
def list():
    qr = Bookmark.select()
    return object_list('bookmarks/index.html', qr)

@bookmarks.route('/add/')
@auth.login_required
def add():
    url = request.args.get('url')
    if url:
        Bookmark.get_or_create(url=url)
    return redirect(url or url_for('bookmarks.list'))

@bookmarks.route('/<pk>/delete/')
@auth.login_required
def delete(pk):
    bookmark = get_object_or_404(Bookmark, id=pk)
    bookmark.delete_instance()
    return redirect(url_for('bookmarks.list'))


class BookmarkResource(RestResource):
    pass

pastebin.py

"""
Simple pastebin
"""

import datetime
import hashlib

from flask import request, redirect, url_for, render_template, Blueprint
from peewee import *
from pygments import formatters, highlight, lexers
from wtfpeewee.orm import model_form

from flaskext.rest import RestResource
from flaskext.utils import get_object_or_404, object_list

from api import api
from app import app, db
from auth import auth


class Paste(db.Model):
    filename = CharField()
    content = TextField()
    highlighted = TextField()
    created_date = DateTimeField(default=datetime.datetime.now)
    sha = CharField()

    class Meta:
        ordering = (('created_date', 'desc'),)

    def __unicode__(self):
        return self.filename

    def save(self):
        self.highlighted = self.highlight()
        self.sha = hashlib.sha1(self.content).hexdigest()
        super(Paste, self).save()

    def highlight(self):
        lexer = None
        try:
            lexer = lexers.get_lexer_for_filename(self.filename)
        except lexers.ClassNotFound:
            lexer = lexers.guess_lexer(self.content)
        formatter = formatters.HtmlFormatter(linenos=True)
        return highlight(self.content, lexer, formatter)


pastebin = Blueprint('pastebin', __name__, static_folder='static', template_folder='templates')

@pastebin.route('/')
@auth.login_required
def list():
    qr = Paste.select()
    return object_list('pastebin/index.html', qr)

@pastebin.route('/add/', methods=['GET', 'POST'])
@auth.login_required
def add():
    Form = model_form(Paste, only=('filename', 'content',))
    if request.method == 'POST':
        form = Form(request.form)
        if form.validate():
            instance = Paste()
            form.populate_obj(instance)
            instance.save()
            return redirect(url_for('pastebin.list'))
    else:
        form = Form()
    return render_template('pastebin/add.html', form=form)

@pastebin.route('/<sha>/')
def detail(sha):
    paste = get_object_or_404(Paste, sha=sha)
    return render_template('pastebin/detail.html', paste=paste)

@pastebin.route('/<sha>/delete/', methods=['GET', 'POST'])
@auth.login_required
def delete(sha):
    paste = get_object_or_404(Paste, sha=sha)
    if request.method == 'POST':
        paste.delete_instance()
        return redirect(url_for('pastebin.list'))
    return render_template('pastebin/delete.html', paste=paste)


class PasteResource(RestResource):
    exclude = ('highlighted',)

pb.py

#!/usr/bin/env python

"""
Command-line tool for posting pastes
"""

# gitconfig
# paste.user / paste.key

# env
# PASTE_USER PASTE_KEY

import datetime
import httplib2
import json
import optparse
import os
import subprocess
import sys


ENDPOINT = '/api/paste/'
URL_TEMPLATE = '/pastebin/%s/'

def get_credentials(options):
    if options.user and options.key:
        return [options.user, options.key]
    elif 'PASTE_USER' in os.environ and 'PASTE_KEY' in os.environ:
        return [os.environ['PASTE_USER'], os.environ['PASTE_KEY']]
    else:
        data = []
        for piece in ('user', 'key'):
            p = subprocess.Popen(['git', 'config', '--global', 'paste.%s' % piece], stdout=subprocess.PIPE)
            data.append(p.communicate()[0].strip())
        return data

def post_file(user, key, filename, content):
    sock = httplib2.Http()
    sock.add_credentials(user, key)
    headers, resp = sock.request(ENDPOINT, 'POST', json.dumps({
        'filename': filename,
        'content': content,
    }))
    if headers['status'] == '200':
        resp_data = json.loads(resp)
        print URL_TEMPLATE % resp_data['sha']
    else:
        print 'Error: %s' % headers['status']
        print resp

def main(user, key, args):
    if args:
        for filename in args:
            fh = open(filename)
            post_file(user, key, os.path.basename(filename), fh.read())
    else:
        now = datetime.datetime.now()
        post_file(user, key, now.strftime('%Y-%m-%d %H:%M:%S'), sys.stdin.read())

if __name__ == '__main__':
    parser = optparse.OptionParser()
    parser.add_option('-u', '--user', dest='user')
    parser.add_option('-k', '--key', dest='key')

    options, args = parser.parse_args()

    user, key = get_credentials(options)
    if not user or not key:
        print 'Error: unable to determine user or api key'
        sys.exit(1)

    main(user, key, args)

pings.py

"""
Ping a list of URLs every 30 minutes
"""


import datetime
import httplib2
import socket
import time

from flask import request, render_template, Blueprint
from peewee import *
from skew.decorators import periodic_command, crontab

from flaskext.rest import RestResource
from flaskext.utils import get_object_or_404, object_list

from api import api
from app import app, db, invoker
from auth import auth


class Site(db.Model):
    url = TextField()
    timeout = IntegerField(default=10)

    def __unicode__(self):
        return self.url

    def check(self):
        ping = Ping(site=self)
        ping.ping()
        ping.save()
        return ping

class Ping(db.Model):
    site = ForeignKeyField(Site, related_name='pings')
    status = IntegerField()
    error_fetching = BooleanField(default=False)
    exception = TextField()
    response_time = FloatField()
    created_date = DateTimeField(default=datetime.datetime.now)

    class Meta:
        ordering = (('created_date', 'desc'),)

    def __unicode__(self):
        return '%s: %s in %s' % (self.site.url, self.status, self.response_time)

    def ping(self):
        try:
            start = time.time()
            sock = httplib2.Http(timeout=self.site.timeout)
            headers, resp = sock.request(self.site.url)
        except (socket.error, AttributeError, httplib2.ServerNotFoundError):
            self.error_fetching = True
            exception = sys.exc_info()[1]
            self.exception = repr(exception)
            self.status = -1
        else:
            self.status = int(headers['status'])
        
        self.response_time = time.time() - start


pings = Blueprint('pings', __name__, template_folder='templates')

@pings.route('/')
@auth.login_required
def list():
    qr = Site.select().order_by('url')
    return object_list('pings/index.html', qr, paginate_by=20)


@periodic_command(invoker, crontab(minute='*/30'))
def ping_sites():
    for site in Site.select():
        site.check()


class SiteResource(RestResource):
    pass

class PingResource(RestResource):
    pass