Skip to content

Instantly share code, notes, and snippets.

View x_endpoint_api_userinfo.py
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)
View select_distinct.py
# Assume these classes are fully defined, including relationships
class User(Base):
pass
class Order(Base):
pass
# Requested: SELECT DISTINCT user.id FROM user, order WHERE user.id = order.user_id
session.query(User.id).select_from(User, Order).filter(User.id == Order.user_id).distinct()
# Produces: SELECT DISTINCT user.id AS user_id FROM user, order WHERE user.id = order.user_id
View tz-sqlalchemy.py
from sqlalchemy.types import DATETIME # optionally import this from your specific dialect
from sqlalchemy.types import TypeDecorator
import datetime
import pytz
class TzDateTime(TypeDecorator):
impl = DATETIME
tz = pytz.utc
def process_bind_param(self, value, dialect):
@inklesspen
inklesspen / likeindex.lisp
Created Mar 29, 2018
download all tumblr likes and store in sqlite db
View likeindex.lisp
(defvar *client*
(make-instance
'north:client
:key "key"
:secret "secret"
:token "token"
:token-secret "secret"
:request-token-uri "https://www.tumblr.com/oauth/request_token"
:authorize-uri "https://www.tumblr.com/oauth/authorize"
:access-token-uri "https://www.tumblr.com/oauth/access_token"))
View reload_eagerly.py
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
inklesspen / __init__.py
Created Oct 8, 2016
ACL Example with Pyramid non-traversal
View __init__.py
from pyramid.config import Configurator
from pyramid.session import SignedCookieSessionFactory
from pyramid.authentication import SessionAuthenticationPolicy
from pyramid.authorization import ACLAuthorizationPolicy
from pyramid.security import Allow, Authenticated, Everyone
class Public(object):
__acl__ = [(Allow, Everyone, 'view')]
View dependency graph.py
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()]
View sqla-code-formatting.md

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(Opportunity.id).filter(Opportunity.is_complete == sa.sql.true(),
                            Opportunity.created_by_id == flask.session.user_id,
                            Opportunity.is_staff_created == sa.sql.false()).count()
View mimir_blogpost_mallows.py
# https://github.com/inklesspen/mimir/blob/fee47910e149094a339a26297324d61568dbadf3/mimir/models/mallows.py
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()
View mimir_blogpost_requestlocals.py
from sqlalchemy import engine_from_config
from sqlalchemy.orm import sessionmaker
from zope.sqlalchemy import register
import transaction
# https://github.com/inklesspen/mimir/blob/fee47910e149094a339a26297324d61568dbadf3/mimir/models/__init__.py#L259-L278
def create_session(request):
sessionmaker = request.registry['db_sessionmaker']
session = sessionmaker()
register(session, transaction_manager=request.tm)
You can’t perform that action at this time.