Skip to content

Instantly share code, notes, and snippets.

@coleifer coleifer/bookmarks.py Secret
Created Oct 4, 2011

Embed
What would you like to do?
Fun stuff I wrote with flask-peewee
"""
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
"""
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',)
#!/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)
"""
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
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.