Skip to content

Instantly share code, notes, and snippets.

import base64
import json
def extract_jwt_claims(req):
header = req.headers.get('X-Endpoint-API-UserInfo')
if header is None:
return None
val = json.loads(base64.urlsafe_b64decode(header))
claims_json = val['claims']
claims = json.loads(claims_json)
# Assume these classes are fully defined, including relationships
class User(Base):
class Order(Base):
# Requested: SELECT DISTINCT FROM user, order WHERE = order.user_id
session.query(, Order).filter( == Order.user_id).distinct()
# Produces: SELECT DISTINCT AS user_id FROM user, order WHERE = order.user_id
from sqlalchemy.types import DATETIME # optionally import this from your specific dialect
from sqlalchemy.types import TypeDecorator
import datetime
import pytz
class TzDateTime(TypeDecorator):
tz = pytz.utc
def process_bind_param(self, value, dialect):
inklesspen / likeindex.lisp
Created Mar 29, 2018
download all tumblr likes and store in sqlite db
View likeindex.lisp
(defvar *client*
:key "key"
:secret "secret"
:token "token"
:token-secret "secret"
:request-token-uri ""
:authorize-uri ""
:access-token-uri ""))
from sqlalchemy import inspect
from sqlalchemy.orm import joinedload
def reload_eagerly(obj, *rels):
state = inspect(obj)
if not state.persistent:
raise ValueError("reload_eagerly can only be used on persistent objects")
query = state.session.query(state.mapper).filter(*[col == val for col, val in zip(state.mapper.primary_key, state.identity)])
for rel in rels:
if rel in state.mapper.all_orm_descriptors:
inklesspen /
Created Oct 8, 2016
ACL Example with Pyramid non-traversal
from pyramid.config import Configurator
from pyramid.session import SignedCookieSessionFactory
from pyramid.authentication import SessionAuthenticationPolicy
from pyramid.authorization import ACLAuthorizationPolicy
from import Allow, Authenticated, Everyone
class Public(object):
__acl__ = [(Allow, Everyone, 'view')]
View dependency
import graphviz
import pkg_resources
pkg_keys = {}
for item in pkg_resources.working_set:
r = item.as_requirement()
pkg_keys[r.key] = r.project_name
depends = {}
for item in pkg_resources.working_set:
depends[item.as_requirement().key] = [r.key for r in item.requires()]

A SQLAlchemy query is a generative object that you keep obtaining a modified copy of with successive method calls. Often, the final method call executes the query and returns a value. Python's built-in line continuation can obscure the structure, so explicitly use \ here.

Instead of this:

    return db_session.query( == sa.sql.true(),
                            Opportunity.created_by_id == flask.session.user_id,
                            Opportunity.is_staff_created == sa.sql.false()).count()
class Writeup(Schema):
id = fields.Integer()
author_slug = fields.String()
writeup_slug = fields.String()
title = fields.String()
status = fields.String()
published = fields.Boolean()
offensive_content = fields.Boolean()
triggery_content = fields.Boolean()
from sqlalchemy import engine_from_config
from sqlalchemy.orm import sessionmaker
from zope.sqlalchemy import register
import transaction
def create_session(request):
sessionmaker = request.registry['db_sessionmaker']
session = sessionmaker()
You can’t perform that action at this time.