cassiacommon

提供了常用的工具类或函数。

SimpleFuture

一个轻量级的异步结果容器，用于在异步代码中等待一个事件完成并获取其结果。

示例

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

轻量级一次性事件同步原语，基于 asyncio.ThreadSafeFlag 实现。

__init__() → SimpleFuture

初始化对象

set_result(code, value)

设置结果并唤醒等待者。只能生效一次，重复调用会被忽略。

Parameters:

• code: 自定义状态码

• value: 需要传递的数据

返回:

无

async wait() → Tuple[code, value]

挂起直到结果就绪，然后返回打包好的数据。

Parameters:

无

返回:

• code: 自定义状态码

• value: 传递的数据

---

AsyncQueue

一个轻量级的异步队列，用于在异步任务之间传递数据。

示例

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

轻量级一次性事件同步原语，基于 asyncio.ThreadSafeFlag 实现。

__init__(max_size=16) → AsyncQueue

初始化对象

put_nowait(item)

非阻塞放入元素。如果队列已满，则直接丢弃，不抛出异常。

Parameters:

• item: 待放入的对象

返回:

无

async get() → Any

取出队首元素。若队列为空，则挂起直到有数据。

Parameters:

无

返回:

最早放入的元素

---

DataCache

一个带大小限制的数据缓存，支持自定义设置 max_bytes 和 hashtable_size。

类型	说明	缺省
max_bytes	总大小限制	256K
hashtable_size	哈希大小	53

示例

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

一个带大小限制的数据缓存，支持自定义设置 max_bytes 和 hashtable_size。

_DATA_CACHE_MAX_MEM

默认最大内存使用量，值为 256 KB (262,144 字节)

_DATA_CACHE_HASHTABLE_SIZE

哈希表默认大小，值为 53

_DATA_CACHE_GET_MAX_CNT

默认最大获取数量，值为 10

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

初始化对象

info()

获取缓存信息

Parameters:

无

返回:

• current_size: 已使用空间

• max_size: 总空间大小

async put(key, data)

异步存储数据到缓存

参数:

• key (str) -- 数据键

• data (any) -- 要存储的数据

返回:

无返回值

async get_next(cnt=_DATA_CACHE_GET_MAX_CNT, flag=False)

异步获取下一批数据

参数:

• cnt (int) -- 获取的数据数量，默认为 _DATA_CACHE_GET_MAX_CNT

• flag (bool) -- 排序标志，True 表示按键顺序，False 表示按时间或队列顺序

返回:

数据列表

返回类型:

list

如果缓存未初始化或没有数据，返回空列表。

async get(key, cnt=_DATA_CACHE_GET_MAX_CNT)

异步根据键获取数据

参数:

• key (str) -- 数据键

• cnt (int) -- 获取的数据数量，默认为 _DATA_CACHE_GET_MAX_CNT

返回:

数据列表

返回类型:

list

如果缓存未初始化或没有对应键的数据，返回空列表。