microsoft--graphrag/packages/graphrag-cache/example_notebooks/custom_cache_example.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "42524e97",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Copyright (c) 2026 Microsoft Corporation.\n",
    "# Licensed under the MIT License."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a7cd6743",
   "metadata": {},
   "source": [
    "## Custom cache example\n",
    "\n",
    "This example demonstrates how to create a custom cache implementation by extending the base Cache class and registering it with the GraphRAG cache system. Once registered, the custom cache can be instantiated through the factory pattern using either CacheConfig or directly via cache_factory, allowing for extensible caching solutions tailored to specific needs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "e7ec8dc1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Value stored in cache for 'my_key':\n",
      "{'k1': 'object to cache'}\n"
     ]
    }
   ],
   "source": [
    "from typing import Any\n",
    "\n",
    "from graphrag_cache import Cache, CacheConfig, create_cache, register_cache\n",
    "\n",
    "\n",
    "class MyCache(Cache):\n",
    "    \"\"\"Custom cache implementation for storing and retrieving cached data.\"\"\"\n",
    "\n",
    "    def __init__(\n",
    "        self,\n",
    "        some_setting: str,\n",
    "        optional_setting: str = \"default setting\",\n",
    "        **kwargs: Any,\n",
    "    ):\n",
    "        # Validate settings and initialize\n",
    "        # View the JsonCache implementation to see how to create a cache that relies on a Storage provider.\n",
    "        self.some_setting = some_setting\n",
    "        self.optional_setting = optional_setting\n",
    "        self._cache: dict[str, Any] = {}\n",
    "        self._child: Cache = self\n",
    "\n",
    "    async def get(self, key: str) -> Any:\n",
    "        \"\"\"Retrieve a value from the cache by key.\"\"\"\n",
    "        return self._cache.get(key)\n",
    "\n",
    "    async def set(self, key: str, value: Any, debug_data: Any = None) -> None:\n",
    "        \"\"\"Store a value in the cache with the specified key.\"\"\"\n",
    "        self._cache[key] = value\n",
    "\n",
    "    async def has(self, key: str) -> bool:\n",
    "        \"\"\"Check if a key exists in the cache.\"\"\"\n",
    "        return key in self._cache\n",
    "\n",
    "    async def delete(self, key: str) -> None:\n",
    "        \"\"\"Remove a key and its value from the cache.\"\"\"\n",
    "        self._cache.pop(key, None)\n",
    "\n",
    "    async def clear(self) -> None:\n",
    "        \"\"\"Clear all items from the cache.\"\"\"\n",
    "        self._cache.clear()\n",
    "\n",
    "    def child(self, name: str) -> Cache:\n",
    "        \"\"\"Create or access a child cache (not implemented in this example).\"\"\"\n",
    "        return self._child\n",
    "\n",
    "\n",
    "register_cache(\"MyCache\", MyCache)\n",
    "\n",
    "\n",
    "async def run():\n",
    "    \"\"\"Demonstrate usage of the custom cache implementation.\"\"\"\n",
    "    cache = create_cache(\n",
    "        CacheConfig(\n",
    "            type=\"MyCache\",\n",
    "            some_setting=\"important setting value\",  # type: ignore\n",
    "        )\n",
    "    )\n",
    "\n",
    "    await cache.set(\"my_key\", {\"k1\": \"object to cache\"})\n",
    "    print(\"Value stored in cache for 'my_key':\")\n",
    "    print(await cache.get(\"my_key\"))\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    await run()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}