### Complete Plugin Example

Source: https://github.com/aio-libs/aiocache/blob/master/docs/plugins.md

Demonstrates using HitMissRatioPlugin, TimingPlugin, and a custom plugin (MyCustomPlugin) with a SimpleMemoryCache. The custom plugin logs pre and post calls for the 'set' command. This example runs a series of get operations to populate cache statistics.

```python
import asyncio
import random
import logging

from aiocache import SimpleMemoryCache
from aiocache.plugins import HitMissRatioPlugin, TimingPlugin, BasePlugin


logger = logging.getLogger(__name__)


class MyCustomPlugin(BasePlugin):

    async def pre_set(self, *args, **kwargs):
        logger.info("I'm the pre_set hook being called with %s %s" % (args, kwargs))

    async def post_set(self, *args, **kwargs):
        logger.info("I'm the post_set hook being called with %s %s" % (args, kwargs))


cache = SimpleMemoryCache(
    plugins=[HitMissRatioPlugin(), TimingPlugin(), MyCustomPlugin()],
    namespace="main")


async def run():
    await cache.set("a", "1")
    await cache.set("b", "2")
    await cache.set("c", "3")
    await cache.set("d", "4")

    possible_keys = ["a", "b", "c", "d", "e", "f"]

    for t in range(1000):
        await cache.get(random.choice(possible_keys))

    assert cache.hit_miss_ratio["hit_ratio"] > 0.5
    assert cache.hit_miss_ratio["total"] == 1000

    assert cache.profiling["get_min"] > 0
    assert cache.profiling["set_min"] > 0
    assert cache.profiling["get_max"] > 0
    assert cache.profiling["set_max"] > 0

    print(cache.hit_miss_ratio)
    print(cache.profiling)


async def test_run():
    await run()
    await cache.delete("a")
    await cache.delete("b")
    await cache.delete("c")
    await cache.delete("d")


if __name__ == "__main__":
    asyncio.run(test_run())
```

--------------------------------

### Install Development Dependencies

Source: https://github.com/aio-libs/aiocache/blob/master/CONTRIBUTING.rst

Install the necessary dependencies for development using the provided Makefile.

```bash
make install-dev
```

--------------------------------

### Basic Cache Usage with SimpleMemoryCache

Source: https://github.com/aio-libs/aiocache/blob/master/docs/index.md

Demonstrates basic cache operations like setting and getting a value using SimpleMemoryCache. Requires asyncio.Runner for execution.

```python
>>> import asyncio
>>> from aiocache import SimpleMemoryCache
>>> cache = SimpleMemoryCache()
>>> with asyncio.Runner() as runner:
>>>     runner.run(cache.set("key", "value"))
True
>>>     runner.run(cache.get("key"))
'value'
```

--------------------------------

### Basic Cache Usage with SimpleMemoryCache

Source: https://github.com/aio-libs/aiocache/blob/master/README.rst

Demonstrates setting and getting a value using SimpleMemoryCache. Ensure you are running this within an asyncio event loop.

```python
>>> import asyncio
>>> from aiocache import SimpleMemoryCache
>>> cache = SimpleMemoryCache()  # Or RedisCache, MemcachedCache...
>>> with asyncio.Runner() as runner:
>>>     runner.run(cache.set('key', 'value'))
True
>>>     runner.run(cache.get('key'))
'value'
```

--------------------------------

### Implement a Custom Compression Serializer

Source: https://github.com/aio-libs/aiocache/blob/master/docs/serializers.md

This example shows how to create a custom serializer that compresses data using zlib before storing it and decompresses it upon retrieval. Set `DEFAULT_ENCODING = None` if your serializer works with bytes.

```python
import asyncio
import zlib

from glide import GlideClientConfiguration, NodeAddress

from aiocache import ValkeyCache
from aiocache.serializers import BaseSerializer


addresses = [NodeAddress("localhost", 6379)]
config = GlideClientConfiguration(addresses=addresses, database_id=0)


class CompressionSerializer(BaseSerializer):

    # This is needed because zlib works with bytes.
    # this way the underlying backend knows how to
    # store/retrieve values
    DEFAULT_ENCODING = None

    def dumps(self, value):
        print("I've received:\n{}".format(value))
        compressed = zlib.compress(value.encode())
        print("But I'm storing:\n{}".format(compressed))
        return compressed

    def loads(self, value):
        print("I've retrieved:\n{}".format(value))
        decompressed = zlib.decompress(value).decode()
        print("But I'm returning:\n{}".format(decompressed))
        return decompressed


async def serializer(cache):
    text = (
        "Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt"
        "ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation"
        "ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in"
        "reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur"
        "sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit"
        "anim id est laborum."
    )
    await cache.set("key", text)
    print("-----------------------------------")
    real_value = await cache.get("key")
    compressed_value = await cache.raw("get", "main:key")
    assert len(compressed_value) < len(real_value.encode())


async def test_serializer():
    async with ValkeyCache(
        config, namespace="main", serializer=CompressionSerializer()
    ) as cache:
        await serializer(cache)
        await cache.delete("key")
        await cache.close()


if __name__ == "__main__":
    asyncio.run(test_serializer())

```

--------------------------------

### Clone aiocache Repository

Source: https://github.com/aio-libs/aiocache/blob/master/CONTRIBUTING.rst

Clone the aiocache repository from GitHub to start contributing.

```bash
git clone git@github.com:argaen/aiocache.git
```

--------------------------------

### Implement a Custom Marshmallow Serializer

Source: https://github.com/aio-libs/aiocache/blob/master/docs/serializers.md

This example demonstrates using marshmallow schemas for serialization and deserialization of complex Python objects. The `RandomModel` and `RandomSchema` classes define the structure and serialization logic.

```python
import random
import string
import asyncio
from typing import Any

from marshmallow import fields, Schema, post_load

from aiocache import SimpleMemoryCache
from aiocache.serializers import BaseSerializer


class RandomModel:
    MY_CONSTANT = "CONSTANT"

    def __init__(self, int_type=None, str_type=None, dict_type=None, list_type=None):
        self.int_type = int_type or random.randint(1, 10)
        self.str_type = str_type or random.choice(string.ascii_lowercase)
        self.dict_type = dict_type or {}
        self.list_type = list_type or []

    def __eq__(self, obj):
        return self.__dict__ == obj.__dict__


class RandomSchema(Schema):
    int_type = fields.Integer()
    str_type = fields.String()
    dict_type = fields.Dict()
    list_type = fields.List(fields.Integer())

    @post_load
    def build_my_type(self, data, **kwargs):
        return RandomModel(**data)

    class Meta:
        strict = True


class MarshmallowSerializer(BaseSerializer):
    def __init__(self, *args: Any, **kwargs: Any):
        super().__init__(*args, **kwargs)
        self.schema = RandomSchema()

    def dumps(self, value: Any) -> str:
        return self.schema.dumps(value)

    def loads(self, value: str) -> Any:
        return self.schema.loads(value)


cache = SimpleMemoryCache(serializer=MarshmallowSerializer(), namespace="main")


async def serializer():
    model = RandomModel()
    await cache.set("key", model)

    result = await cache.get("key")

    assert result.int_type == model.int_type
    assert result.str_type == model.str_type
    assert result.dict_type == model.dict_type
    assert result.list_type == model.list_type


async def test_serializer():
    await serializer()
    await cache.delete("key")


if __name__ == "__main__":
    asyncio.run(test_serializer())

```

--------------------------------

### Using aiocache's Cached Decorator with Redis

Source: https://github.com/aio-libs/aiocache/blob/master/README.rst

Shows how to use the @cached decorator to automatically cache the results of an async function. This example uses Redis as the backend and PickleSerializer to store Python objects. Ensure Redis is running and accessible.

```python
import asyncio

from collections import namedtuple

from aiocache import RedisCache, cached
from aiocache.serializers import PickleSerializer
# With this we can store python objects in backends like Redis!

Result = namedtuple('Result', "content, status")
redis_client = redis.Redis(host="127.0.0.1", port=6379)
redis_cache = RedisCache(redis_client, namespace="main")


@cached(redis_cache, key="key", serializer=PickleSerializer(), port=6379, namespace="main")
async def cached_call():
    print("Sleeping for three seconds zzzz.....")
    await asyncio.sleep(3)
    return Result("content", 200)


async def run():
    async with redis_client, redis_cache:
        await cached_call()
        await cached_call()
        await cached_call()
        await redis_cache.delete("key")

if __name__ == "__main__":
    asyncio.run(run())
```

--------------------------------

### Use @cached decorator for asynchronous functions

Source: https://github.com/aio-libs/aiocache/blob/master/docs/decorators.md

Use the @cached decorator to cache the results of an asynchronous function. Do not use it with synchronous functions as it may lead to unexpected behavior. Ensure the cache object is properly initialized and managed, for example, using an async with statement.

```python
import asyncio
from collections import namedtuple

from glide import GlideClientConfiguration, NodeAddress

from aiocache import cached
from aiocache import ValkeyCache
from aiocache.serializers import PickleSerializer

Result = namedtuple("Result", "content, status")

addresses = [NodeAddress("localhost", 6379)]
config = GlideClientConfiguration(addresses=addresses, database_id=0)
cache = ValkeyCache(config=config, namespace="main", serializer=PickleSerializer())


@cached(cache, ttl=10, key_builder=lambda *args, **kw: "key")
async def cached_call():
    return Result("content", 200)


async def test_cached():
    async with cache:
        await cached_call()
        exists = await cache.exists("key")
        assert exists is True
        await cache.delete("key")


if __name__ == "__main__":
    asyncio.run(test_cached())
```

--------------------------------

### Build and Serve Documentation

Source: https://github.com/aio-libs/aiocache/blob/master/CONTRIBUTING.rst

Build the project documentation using Sphinx and serve it locally for review.

```bash
sphinx-autobuild docs/ docs/_build/html/
```

--------------------------------

### Initialize Cache with Plugins

Source: https://github.com/aio-libs/aiocache/blob/master/docs/plugins.md

Add plugins to a cache instance during its creation. Ensure the plugin classes are imported.

```python
>>> from aiocache import SimpleMemoryCache
>>> from aiocache.plugins import TimingPlugin
cache = SimpleMemoryCache(plugins=[HitMissRatioPlugin()])
cache.plugins += [TimingPlugin()]
```

--------------------------------

### Run All Tests

Source: https://github.com/aio-libs/aiocache/blob/master/CONTRIBUTING.rst

Execute all tests to ensure code quality and stability. Requires Docker and docker-compose for acceptance and functional tests.

```bash
make test
```

--------------------------------

### Initialize Cache with PickleSerializer

Source: https://github.com/aio-libs/aiocache/blob/master/docs/serializers.md

Instantiate a SimpleMemoryCache and attach a PickleSerializer to it. This is useful when you need to store Python objects that require serialization.

```python
from aiocache import SimpleMemoryCache
from aiocache.serializers import PickleSerializer

cache = SimpleMemoryCache(serializer=PickleSerializer())
```

--------------------------------

### Format Code

Source: https://github.com/aio-libs/aiocache/blob/master/CONTRIBUTING.rst

Automatically format the code to comply with project standards.

```bash
make format
```

--------------------------------

### Run Unit Tests

Source: https://github.com/aio-libs/aiocache/blob/master/CONTRIBUTING.rst

Run only the unit tests for faster iteration during development.

```bash
make unit
```

--------------------------------

### Mocking BaseCache for Testing

Source: https://github.com/aio-libs/aiocache/blob/master/docs/testing.md

Use this snippet to mock the BaseCache for testing purposes. Ensure you pass the BaseCache as the spec for the mock object.

```python
import asyncio
from unittest.mock import MagicMock

from aiocache.base import BaseCache


async def main():
    mocked_cache = MagicMock(spec=BaseCache)
    mocked_cache.get.return_value = "world"
    print(await mocked_cache.get("hello"))


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Lint Code

Source: https://github.com/aio-libs/aiocache/blob/master/CONTRIBUTING.rst

Ensure the code adheres to the project's syntax standards.

```bash
make lint
```

--------------------------------

### Use @multi_cached decorator with attribute for multiple keys

Source: https://github.com/aio-libs/aiocache/blob/master/docs/decorators.md

The @multi_cached decorator caches results for multiple keys, specified by `keys_from_attr`. This is useful when a function returns a dictionary where keys correspond to cache keys. Ensure the cache object is properly initialized and managed.

```python
import asyncio

from glide import GlideClientConfiguration, NodeAddress

from aiocache import multi_cached
from aiocache import ValkeyCache

DICT = {"a": "Z", "b": "Y", "c": "X", "d": "W"}

addresses = [NodeAddress("localhost", 6379)]
config = GlideClientConfiguration(addresses=addresses, database_id=0)
cache = ValkeyCache(config=config, namespace="main")


@multi_cached(cache, keys_from_attr="ids")
async def multi_cached_ids(ids=None):
    return {id_: DICT[id_] for id_ in ids}


@multi_cached(cache, keys_from_attr="keys")
async def multi_cached_keys(keys=None):
    return {id_: DICT[id_] for id_ in keys}


async def test_multi_cached():
    async with cache:
        await multi_cached_ids(ids=("a", "b"))
        await multi_cached_ids(ids=("a", "c"))
        await multi_cached_keys(keys=("d",))

        assert await cache.exists("a")
        assert await cache.exists("b")
        assert await cache.exists("c")
        assert await cache.exists("d")

        await cache.delete("a")
        await cache.delete("b")
        await cache.delete("c")
        await cache.delete("d")


if __name__ == "__main__":
    asyncio.run(test_multi_cached())
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.