Skip to main content

Cache integration with sqlalchemy.

Project description

Run test

make unittest

Installation / Rquirements

pip intall ecache

Usage

With Flask Integrate

from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy

from ecache.ext.flask_cache import CacheableMixin, query_callable, regions

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
app.debug = True
db = SQLAlchemy(app)

class User(db.Model, CacheableMixin):
    """Default backend is redis and expiration time is 1 hour, default
    region name is `default`, you can override this:

        cache_regions = your_regions
        cache_label = your_label
    """

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String)

@app.route('/users')
def all_users():
    """Result will try to get from cache first. load from db if cache miss.
    """
    users = [user.to_dict() for user in User.cache.filter()]
    return jsonify(users=users)


@app.route('/users/<int:user_id')
def view_user(user_id):
    """Result will try to get from cache first. load from db if cache miss.
    """
    user = User.cache.get(user_id)
    return jsonify(user.to_dict())

More detail see example

With Pure SQLAlchemy model Integrate

# -*- coding: utf-8 -*-

import redis
from sqlalchemy import (Column, Integer, String, SmallInteger)

from ecache.core import cache_mixin
from ecache.db import db_manager, model_base

# alsosee :class:`ecache.db.DBManager`
DBSession = db_manager.get_session('test')
cache_client = redis.StrictRedis()
CacheMixin = cache_mixin()
DeclarativeBase = model_base()


class TodoListModel(DeclarativeBase, CacheMixin):
    __tablename__ == 'todo_list'
    TABLE_CACHE_EXPIRATION_TIME = 3600

    id = Column(Integer, primary_key=True)
    title = Column(String, default='')
    is_done = Column(SmallInteger, default=0)

    @classmethod
    def get_todo(cls, todo_id):
        todo = cls.get(todo_id)  # `cls.get` inherited from `CacheMixin`
        return todo

    @classmethod
    def add(cls, title):
        todo = cls(title=title)
        s = DBSession()
        s.add(todo)
        s.commit()

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Filename, size & hash SHA256 hash help File type Python version Upload date
ecache-0.1.2-py2-none-any.whl (16.7 kB) Copy SHA256 hash SHA256 Wheel 2.7
ecache-0.1.2.tar.gz (11.9 kB) Copy SHA256 hash SHA256 Source None

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page