Metadata-Version: 2.1
Name: mapgraph
Version: 0.2.3
Summary: 一个强大的对象控制工具, 覆盖了配置管理、状态管理、上下文管理和生命周期管理。
Author-Email: luxuncang <luxuncang@qq.com>
License: MIT
Requires-Python: >=3.10
Requires-Dist: python-dotenv>=1.0.1
Requires-Dist: launart>=0.8.2
Requires-Dist: typing-tool>=0.1.2
Requires-Dist: networkx>=3.4.2
Provides-Extra: aiohttp
Requires-Dist: aiohttp>=3.10.5; extra == "aiohttp"
Provides-Extra: httpx
Requires-Dist: httpx>=0.27.2; extra == "httpx"
Provides-Extra: uvicorn
Requires-Dist: uvicorn>=0.30.6; extra == "uvicorn"
Provides-Extra: starlette
Requires-Dist: starlette>=0.38.4; extra == "starlette"
Provides-Extra: tortoise
Requires-Dist: tortoise-orm>=0.21.6; extra == "tortoise"
Description-Content-Type: text/markdown

<div align="center">

# Mapgraph

_`mapgraph` 提供了强大的对象控制工具, 覆盖了配置管理、状态管理、上下文管理和生命周期管理。_

> 一阴一阳之谓道

 [![CodeFactor](https://www.codefactor.io/repository/github/LaciaProject/mapgraph/badge)](https://www.codefactor.io/repository/github/LaciaProject/mapgraph)
 [![GitHub](https://img.shields.io/github/license/LaciaProject/mapgraph)](https://github.com/LaciaProject/mapgraph/blob/master/LICENSE)
 [![CodeQL](https://github.com/LaciaProject/mapgraph/workflows/CodeQL/badge.svg)](https://github.com/LaciaProject/mapgraph/blob/master/.github/workflows/codeql.yml)

</div>

## 功能

* 配置管理
  * 基于类型声明的配置
  * 支持实时变更
* 上下文管理
  * 基于描述器的依赖注入
  * 优雅的上下文切换
* 生命周期管理
  * 服务编排
* 状态管理

## 安装

```sh
pip install mapgraph
```

Or

```sh
pdm add mapgraph
```

## 入门指南

### 上下文管理

```python
from mapgraph.context import InstanceContext
```

`InstanceContext` 实例上下文, 顾名思义是管理 `对象` 的抽象, 它通过两个维度对 `对象` 进行管理, 即 `名义类型` 与 `值`。

* `名义类型` 可以理解为 `变量名` 或者是 `变量类型`, 总之它在当前实例上下文是唯一的 `KEY`
* `值` 很好理解，即是某个被管理的 `对象`

```python
from mapgraph.globals import GLOBAL_INSTANCE_CONTEXT
```

`GLOBAL_INSTANCE_CONTEXT` 全局实例上下文, 即最顶层的命名空间, 本质也是一个 `InstanceContext`。

#### 基本使用

```python
from typing import NewType

from mapgraph.globals import GLOBAL_INSTANCE_CONTEXT
from mapgraph.instance_of import get_instance

MYINT = NewType("MYINT", int)

GLOBAL_INSTANCE_CONTEXT.store(123, "test")
# GLOBAL_INSTANCE_CONTEXT.store({int: 123, str: "test"}) 
GLOBAL_INSTANCE_CONTEXT.store({list[int]: [1,2,3]})
GLOBAL_INSTANCE_CONTEXT.store({MYINT: -123})

print(get_instance(int)) # 123
print(get_instance(MYINT)) # 123
print(get_instance(MYINT, is_key=True)) # -123
print(get_instance(str)) # test
print(get_instance(list[int])) # [1,2,3]
```

* `get_instance` 可以通过类型获取 `InstanceContext` 中的值, 当 `is_key=True` 为 `名义类型` 匹配, 反之为 `值` 匹配
* `.store` 拥有可变位置参数 `collection_or_targets` 类型为 `Mapping[type, Any] | Any`
* `.store(123, "test")` 等价于 `.store({int: 123, str: "test"})`

> 因为运行时类型检查器拥有的局限性，所以分别引入了 `名义类型` 与 `值`，它们相互补充

#### 多级实例上下文

```python
from mapgraph.context import InstanceContext
from mapgraph.globals import GLOBAL_INSTANCE_CONTEXT
from mapgraph.instance_of import get_instance

loacl_context = InstanceContext()

GLOBAL_INSTANCE_CONTEXT.store(123, {list[int]: [1,2,3]})
loacl_context.store("test", -123)

print(get_instance(int)) # 123
# print(get_instance(str)) # ValueError: <class 'str'>

with loacl_context.scope():
    print(get_instance(str)) # test
    print(get_instance(int)) # -123
    print(get_instance(list[int])) # [1,2,3]

with loacl_context.scope(inherit=False):
    print(get_instance(str)) # test
    print(get_instance(int)) # -123
    print(get_instance(list[int])) # ValueError: <class 'list[int]'>
```

> `.scope` 拥有关键字参数 `inherit`, 表示是否继承上一级 `InstanceContext`

#### 描述器支持

```python
from typing import NewType

from httpx import AsyncClient
from mapgraph.globals import GLOBAL_INSTANCE_CONTEXT
from mapgraph.instance_of import get_instance, InstanceOf, InstanceOfK, InstanceOfV

OPENAI_API_KEY = NewType("OPENAI_API_KEY", str)

class Test:
    client: InstanceOfV[AsyncClient] = InstanceOf(AsyncClient)
    openai_api_key: InstanceOfK[OPENAI_API_KEY] = InstanceOf(OPENAI_API_KEY, is_key=True)

GLOBAL_INSTANCE_CONTEXT.store({OPENAI_API_KEY: "test"}, AsyncClient())

test = Test()

print(test.client) # <httpx.AsyncClient object at 0x7f7f7f7f7f7f>
print(test.openai_api_key) # test
```

`InstanceOf` 是一个函数会返两种描述器 `InstanceOfV[T]` 与 `InstanceOfK[T]`, 很好理解 前者被声明为通过 `值` 匹配, 后者被声明通过 `名义类型` 匹配。

#### 描述器读取

```python
from typing import NewType

from httpx import AsyncClient
from mapgraph.context import InstanceContext
from mapgraph.globals import GLOBAL_INSTANCE_CONTEXT
from mapgraph.instance_of import get_instance, InstanceOf, InstanceOfK, InstanceOfV

OPENAI_API_KEY = NewType("OPENAI_API_KEY", str)
BASE_URL = NewType("BASE_URL", str)

loacl_context = InstanceContext()

class Test:
    client: InstanceOfV[AsyncClient] = InstanceOf(AsyncClient)
    openai_api_key: InstanceOfK[OPENAI_API_KEY] = InstanceOf(OPENAI_API_KEY, is_key=True)
    base_url: InstanceOfK[BASE_URL] = InstanceOf(BASE_URL, is_key=True)

GLOBAL_INSTANCE_CONTEXT.store({OPENAI_API_KEY: "sk-proj-xxx", BASE_URL: "https://api.openai.com"}, AsyncClient())
loacl_context.store({BASE_URL: "https://api.xxxmy.com"})

test = Test()

print(test.client) # <httpx.AsyncClient object at 0x7f7f7f7f7f7f>
print(test.openai_api_key) # sk-proj-xxx
print(test.base_url) # https://api.openai.com

with loacl_context.scope():
    print(test.base_url) # https://api.xxxmy.com
```

#### 描述器写入

```python
from typing import NewType

from httpx import AsyncClient
from mapgraph.context import InstanceContext
from mapgraph.globals import GLOBAL_INSTANCE_CONTEXT
from mapgraph.instance_of import get_instance, InstanceOf, InstanceOfK, InstanceOfV

OPENAI_API_KEY = NewType("OPENAI_API_KEY", str)
BASE_URL = NewType("BASE_URL", str)

loacl_context = InstanceContext()

class Test:
    client: InstanceOfV[AsyncClient] = InstanceOf(AsyncClient)
    base_url: InstanceOfK[BASE_URL] = InstanceOf(BASE_URL, is_key=True)


GLOBAL_INSTANCE_CONTEXT.store(
    {BASE_URL: "https://api.openai.com"}, AsyncClient()
)
loacl_context.store({BASE_URL: "https://api.xxxmy.com"})

test = Test()

print(id(test.client))  # 139921911645712
print(test.base_url)  # https://api.openai.com

with loacl_context.scope():
    print(test.base_url)  # https://api.xxxmy.com
    print(id(test.client))  # 139921911645712
    test.base_url = {BASE_URL: BASE_URL("https://api.yyymy.com")}
    test.client = AsyncClient()
    print(test.base_url)  # https://api.yyymy.com
    print(id(test.client))  # 140043844997072

print(test.base_url)  # https://api.openai.com
print(id(test.client))  # 139921911645712
```

* 当描述器为 `InstanceOfK[T]` 时, 设置值需要一个 `Mapping[Type[T], T]`
* 当描述器为 `InstanceOfV[T]` 时, 设置值需要为 `T`

### 配置管理

**.env**

```.env
OPENAI_API_KEY = "sk-proj-xxx"
BASE_URL = "https://api.openai.com"
MYINT = 123
MYDICT = {"a": 1}
MYLIST = [1, 2, 3]
```

```python
from typing import NewType

from mapgraph.config import load_env
from mapgraph.instance_of import InstanceOf

OPENAI_API_KEY = NewType("OPENAI_API_KEY", str)
BASE_URL = NewType("BASE_URL", str)
MYINT = NewType("MYINT", int)
MYLIST = NewType("MYLIST", list[int])
MYDICT = NewType("MYDICT", dict[str, int])
MYBOOL = NewType("MYBOOL", bool)

class ConfigData:
    a = InstanceOf(MYINT, is_key=True)
    b = InstanceOf(MYLIST, is_key=True)
    c = InstanceOf(MYDICT, is_key=True)
    d = InstanceOf(MYBOOL, is_key=True)

class Config:
    api_key = InstanceOf(OPENAI_API_KEY, is_key=True)
    base_url = InstanceOf(BASE_URL, is_key=True)
    data = ConfigData()


load_env(
    envs=(OPENAI_API_KEY, BASE_URL, MYINT, MYLIST, MYDICT), 
    envs_kv={MYBOOL: False}, # 设置默认值
    dotenv_path=".env"
)

config = Config()

print(config.api_key) # sk-proj-xxx
print(config.base_url) # https://api.openai.com
print(config.data.a) # 123
print(config.data.b) # [1, 2, 3]
print(config.data.c) # {'a': 1}
print(config.data.d) # False
```

> 最佳实践是控制所需的最细的粒度, 当配置较为复杂可以进行嵌套

### 状态管理

依赖于 [statv](https://github.com/GraiaProject/statv) [文档](https://graia.cn/other/statv/)

### 生命周期管理

目前可以使用两种实现

* `mapgraph.lifecycle` 提供了 `launch_services` 服务自动编排
* 依赖于 [launart](https://github.com/GraiaProject/launart) [文档](https://graia.cn/other/launart/)