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.

Source Distribution

ecache-0.1.2.tar.gz (11.9 kB view hashes)

Uploaded Source

Built Distribution

ecache-0.1.2-py2-none-any.whl (16.7 kB view hashes)

Uploaded Python 2

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page