cassiacommon

Provides commonly used utility classes and functions.

SimpleFuture

A lightweight asynchronous result container used to wait for an event to complete and retrieve its result in asynchronous code.

Example

import asyncio
import cassiacommon


async def main():
    print("start simple future")
    fut = cassiacommon.SimpleFuture()

    async def producer():
        print("producer sleep 3 seconds...")
        await asyncio.sleep(3)
        fut.set_result(200, "OK")
        print("producer ok")

    asyncio.create_task(producer())
    code, value = await fut.wait()
    print("simple future done:", code, value)


asyncio.run(main())

API

class SimpleFuture

A lightweight, one-time event synchronization primitive implemented using asyncio.ThreadSafeFlag.

__init__() → SimpleFuture

Initialize the object

set_result(code, value)

Sets the result and wakes up any waiters. Can only take effect once; subsequent calls are ignored.

Parameters:

• code: Custom status code

• value: Data to be passed

Returns:

None

async wait() → Tuple[code, value]

Suspends execution until the result is ready, then returns the packaged data.

Parameters:

None

Returns:

• code: Custom status code

• value: Data to be passed

---

AsyncQueue

A lightweight asynchronous queue used to pass data between asynchronous tasks.

Example

import asyncio
import cassiacommon


async def main():
    print("start async queue")

    q = cassiacommon.AsyncQueue(max_size=4)
    q.put_nowait("a")
    q.put_nowait("b")
    print(await q.get())
    print(await q.get())

    print("async queue done")


asyncio.run(main())

API

class AsyncQueue

A lightweight, one-time event synchronization primitive implemented using asyncio.ThreadSafeFlag.

__init__(max_size=16) → AsyncQueue

Initialize the object

put_nowait(item)

Puts an element into the queue without blocking. If the queue is full, the element is discarded without raising an exception.

Parameters:

• item: Object to be enqueued

Returns:

None

async get() → Any

Retrieves the element at the front of the queue. If the queue is empty, suspends execution until data becomes available.

Parameters:

None

Returns:

The earliest enqueued element

---

DataCache

A data cache with size limits, supporting custom settings for max_bytes and hashtable_size.

Type	Description	Default
max_bytes	Max Size	256K
hashtable_size	Hash Size	53

Example

import asyncio
import cassiacommon


async def main():
    print("===================")
    print("data cache start")

    cache = cassiacommon.DataCache()

    cache_info = cache.info()
    print(f"cache info init status: {cache_info}")

    # add sample data
    await cache.put("z", {"z_key1": "z_value1"})
    await cache.put("z", {"z_key2": "z_value2"})
    await cache.put("z", [{"z_key3": "z_value3"}])
    
    await cache.put("a", {"a_key1": "a_value1"})
    await cache.put("a", {"a_key2": "a_value2"})
    await cache.put("a", [{"a_key3": "a_value3"}])
    
    # current cache info
    cache_info = cache.info()
    print(f"cache info after put data status: {cache_info}")

    # get data by key
    user_data = await cache.get("a", cnt=1)
    print(f"a data: {user_data}")

    # get no exist key
    noexistent_data = await cache.get("nonexistent:key")
    print(f"no exist data: {noexistent_data}")

    # get data by key order
    key_ordered_data = await cache.get_next(cnt=1, flag=True)
    print(f"key ordered data: {key_ordered_data}")

    # get data by time order
    time_ordered_data = await cache.get_next(cnt=1, flag=False)
    print(f"time ordered data: {time_ordered_data}")

    cache.clear()
    print(f"cache clear ok")
    print(cache.info())

    print("data cache done")


asyncio.run(main())

API

class DataCache

A data cache with size limits, supporting custom settings for max_bytes and hashtable_size.

_DATA_CACHE_MAX_MEM

The default maximum memory usage is 256 KB (262,144 bytes).

_DATA_CACHE_HASHTABLE_SIZE

The default hashtable size is 53.

_DATA_CACHE_GET_MAX_CNT

The default maximum fetch count is 10.

__init__(self, max_bytes=_DATA_CACHE_MAX_MEM, hashtable_size=_DATA_CACHE_HASHTABLE_SIZE) → DataCache

Initialize the object

info()

Get cache information

Parameters:

None

Returns:

• current_size: Used space

• max_size: Total space size

async put(key, data)

Asynchronously store data to cache

Parameters:

• key (str) – Data Key

• data (any) – Data to be stored

Returns:

Return

async get_next(cnt=_DATA_CACHE_GET_MAX_CNT, flag=False)

Asynchronously fetch the next batch of data

Parameters:

• cnt (int) – The number of data items to fetch, defaulting to _DATA_CACHE_GET_MAX_CNT

• flag (bool) – Sort flag, True indicates key order, False indicates time or queue order

Returns:

Data list

Return type:

list

If the cache is not initialized or contains no data, return an empty list.

async get(key, cnt=_DATA_CACHE_GET_MAX_CNT)

Asynchronously fetch data by key.

Parameters:

• key (str) – Data Key

• cnt (int) – The number of data items to fetch, defaulting to _DATA_CACHE_GET_MAX_CNT

Returns:

Data list

Return type:

list

If the cache is not initialized or no data exists for the corresponding key, return an empty list.