Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,14 @@

## 项目定位

sqlmodel-nexus:从 SQLModel 类自动生成 GraphQL API,并提供 Core API 模式构建用例响应的 Python 库。
- **GraphQL 模式**:SDL 自动生成 + DataLoader 批量关系加载 + MCP 服务集成
- **Core API 模式**:DefineSubset DTO + ErManager + Resolver 模型驱动的响应构建
sqlmodel-nexus:从 SQLModel 实体模型出发,自动生成 GraphQL / REST / MCP 三种 API 的 Python 库。

核心理念:定义一次数据模型,获得三种 API 输出,N+1 查询自动防护。

三条使用路径:
- **GraphQL 模式**:在实体上写 `@query`/`@mutation` → 自动生成 SDL → GraphQLHandler 执行查询
- **Core API 模式**:`DefineSubset` DTO → `ErManager` + `Resolver` → 构建 REST 响应 / 业务 DTO
- **UseCase 模式**:`UseCaseService` 业务类 → 同时输出 MCP(AI Agent)和 FastAPI(REST)

## 技术栈

Expand Down
4 changes: 4 additions & 0 deletions docs/advanced/mcp_service.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@

Expose SQLModel APIs to AI agents — create an MCP service with a single line of code.

> **Prerequisites**: SQLModel entities with `@query`/`@mutation` methods and a `session_factory`. This path covers the **GraphQL→MCP** bridge — for the **UseCase→MCP** path see [UseCase Service](./use_case_service.md).
>
> **Live demo**: [`demo/mcp_server_simple.py`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/mcp_server_simple.py) (single-app) and [`demo/mcp_server.py`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/mcp_server.py) (multi-app).

## Installation

```bash
Expand Down
4 changes: 4 additions & 0 deletions docs/advanced/use_case_fastapi.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@

Using the same UseCaseService class in FastAPI — routes are thin wrappers, business logic stays in the service.

> **Prerequisites**: [UseCase Service](./use_case_service.md). You should have a UseCaseService subclass with `@query`/`@mutation` methods before wiring it into FastAPI.
>
> **Prerequisite demo**: [`demo/use_case/mcp_server.py`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/use_case/mcp_server.py) — defines the services. [`demo/use_case/fastapi.py`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/use_case/fastapi.py) — the same services as REST.

## Route Definitions

```python
Expand Down
4 changes: 4 additions & 0 deletions docs/advanced/use_case_service.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@

UseCaseService lets you define business logic as service classes that serve both MCP and web frameworks — one definition, two presentations.

> **Prerequisites**: [Core API Mode](../guide/core_api.md) — ErManager + DefineSubset + Resolver. UseCaseService wraps Core API DTOs into business methods.
>
> **Live demo**: See [`demo/use_case/mcp_server.py`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/use_case/mcp_server.py) for a working MCP server — and [`demo/use_case/fastapi.py`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/use_case/fastapi.py) for the same services exposed as REST.

## Design Philosophy

```
Expand Down
2 changes: 2 additions & 0 deletions docs/advanced/voyager.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

sqlmodel-nexus includes a built-in Voyager module that provides interactive UseCase service graphs and ER entity relationship visualization.

> **Prerequisites**: [UseCase Service](./use_case_service.md) (for service graph) and optionally [Core API Mode](../guide/core_api.md) (for ER diagram integration via `er_manager`).

## create_use_case_voyager

```python
Expand Down
58 changes: 19 additions & 39 deletions docs/guide/core_api.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,11 @@

The Core API mode is for scenarios beyond GraphQL — FastAPI REST endpoints, service layer response assembly, or any use-case DTO. Same DataLoader batch loading, same N+1 prevention.

Core concepts progress in order: **Implicit auto-loading → resolve_\* → post_\* → Cross-layer data flow**.
> **Prerequisites**: SQLModel entities defined with `Relationship` / `Field(foreign_key=...)`.
>
> **Live demo**: See [`demo/core_api/`](https://github.com/allmonday/sqlmodel-nexus/tree/master/demo/core_api) for a complete REST server with Sprint/Task/User models. The DTOs in [`dtos.py`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/core_api/dtos.py) progress through 5 levels of complexity.

Core concepts progress in order: **Implicit auto-loading → resolve\_\* → post\_\* → Cross-layer data flow**.

## Step 1: DefineSubset + Implicit Auto-Loading

Expand All @@ -24,6 +28,8 @@ class SprintDTO(DefineSubset):
tasks: list[TaskDTO] = [] # Name matches Sprint.tasks relationship → auto-loaded
```

> ⬆ This is **Level 2** in the demo's [`dtos.py`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/core_api/dtos.py#L30).

## ErManager Initialization

```python
Expand All @@ -48,49 +54,23 @@ app = FastAPI()
async def get_sprints():
async with async_session() as session:
sprints = (await session.exec(select(Sprint))).all()
dtos = [SprintDTO(id=s.id, name=s.name) for s in sprints]
return await Resolver().resolve(dtos)
dtos = [SprintDTO(**s.model_dump()) for s in sprints]
return await Resolver().resolve(dtos) # tasks + owner auto-loaded
```

## Four Conditions for Implicit Auto-Loading

The Resolver automatically loads relationship fields (no need to write `resolve_*`) when all conditions are met:

1. The field has no corresponding `resolve_*` method
2. The field is an extra field (not in the `__subset__` definition)
3. The field name matches a registered ORM/custom relationship
4. The field type is a BaseModel DTO compatible with the relationship target entity
> **Tip**: For more efficient field-selective queries, use `build_dto_select(SprintDTO)` — it generates a SQLAlchemy select() that only fetches the fields declared in `__subset__`. See [`demo/core_api/dtos.py` → `TaskService.list_tasks`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/use_case/mcp_server.py#L63).

## DefineSubset Rules
## How Auto-Loading Works

- `__subset__` accepts a tuple `(Entity, ('field1', 'field2'))`
- FK fields (e.g., `owner_id`) are automatically hidden from serialization output, but remain internally accessible in `resolve_*`
- Relationship fields are declared in the class body (not in `__subset__`), and must use DTO types

## How It Works

```
SprintDTO(id=1, name="Sprint 1")
→ Implicit auto-load: tasks → [TaskDTO(...), TaskDTO(...)]
→ Each TaskDTO: Implicit auto-load: owner → UserDTO(...)
→ Result: complete nested response tree
```
Implicit auto-loading triggers when **all** conditions are true:

Each relationship executes only one DataLoader query, regardless of how many Sprints or Tasks exist.

## DTO Type Constraint

```python
# Wrong — direct use of SQLModel entity is prohibited
class TaskDTO(DefineSubset):
owner: User | None = None # TypeError!

# Correct — use DTO type
class TaskDTO(DefineSubset):
owner: UserDTO | None = None # OK
```
1. The field has no corresponding `resolve_*` method
2. The field is an extra field (not in `__subset__` fields)
3. The field name matches a registered ORM or custom relationship
4. The field type is a DefineSubset DTO compatible with the relationship target

## Next Steps

- [Core API Advanced](./core_api_advanced.md) — resolve_*/post_*/cross-layer data flow
- [Custom Relationships](./custom_relationship.md) — Non-ORM relationship declarations
- [Core API Advanced](./core_api_advanced.md) — resolve_*, post_*, cross-layer data flow
- [Custom Relationships](./custom_relationship.md) — non-ORM relationships
- [UseCase Service](../advanced/use_case_service.md) — wrap Core API DTOs in business services
92 changes: 55 additions & 37 deletions docs/guide/core_api_advanced.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@

When implicit auto-loading is not enough, Core API provides three progressive capabilities: `resolve_*` for custom loading, `post_*` for derived field computation, and cross-layer data flow.

> **Prerequisites**: [Core API Mode](./core_api.md) — specifically DefineSubset, ErManager, and implicit auto-loading.
>
> **Live demo**: The advanced concepts below correspond to Levels 3–5 in [`demo/core_api/dtos.py`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/core_api/dtos.py).

## resolve_*: Custom Loading

When the field name doesn't match a relationship, or custom logic is needed, use `resolve_*`:
Expand Down Expand Up @@ -58,6 +62,8 @@ class SprintDTO(DefineSubset):
return sorted({t.owner.name for t in self.tasks if t.owner})
```

> ⬆ This is **Level 3** in the demo: [`SprintSummary`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/core_api/dtos.py#L53).

Execution order:

1. Implicit auto-loading → `tasks` populated with TaskDTO list
Expand All @@ -78,60 +84,72 @@ Execution order:

Use when parent and child nodes need cross-layer collaboration. Only necessary when the tree structure truly matters.

### ExposeAs: Ancestor → Descendant
**Key tools**:

```python
from typing import Annotated
from sqlmodel_nexus import ExposeAs
- `ExposeAs`: Parent exposes a value to all descendants via `ancestor_context`
- `SendTo`: Child sends a value upward to ancestor Collector
- `Collector`: Aggregates all values sent via SendTo

class SprintDTO(DefineSubset):
__subset__ = (Sprint, ("id", "name"))
name: Annotated[str, ExposeAs('sprint_name')] # Expose to descendants
tasks: list[TaskDTO] = []
```
```python
from sqlmodel_nexus import Collector, DefineSubset, SubsetConfig

### SendTo + Collector: Descendant → Ancestor
class TaskDTO(DefineSubset):
__subset__ = SubsetConfig(
kls=Task, fields=["id", "title"],
send_to=[("owner", "contributors")], # Send owner to ancestor collector
)
owner: UserDTO | None = None

```python
from sqlmodel_nexus import SendTo, Collector
def post_full_title(self, ancestor_context=None):
sprint_name = (ancestor_context or {}).get("sprint_name", "unknown")
return f"{sprint_name} / {self.title}"

class SprintDTO(DefineSubset):
__subset__ = (Sprint, ("id", "name"))
name: Annotated[str, ExposeAs('sprint_name')]
__subset__ = SubsetConfig(
kls=Sprint, fields=["id", "name"],
expose_as=[("name", "sprint_name")], # Expose name to descendants
)
tasks: list[TaskDTO] = []
contributors: list[UserDTO] = []

def post_contributors(self, collector=Collector('contributors')):
return collector.values() # Collect values sent by descendants

class TaskDTO(DefineSubset):
__subset__ = (Task, ("id", "title", "owner_id"))
owner: Annotated[UserDTO | None, SendTo('contributors')] = None # Send to ancestor
full_title: str = ""

def post_full_title(self, ancestor_context):
return f"{ancestor_context['sprint_name']} / {self.title}"
def post_contributors(self, collector=Collector("contributors")):
return collector.values()
```

Applicable scenarios:
> ⬆ This is **Level 4** in the demo: [`SprintDetail`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/core_api/dtos.py#L107).

- Child nodes need ancestor context (sprint name, permission info, tenant configuration)
- Parent nodes need to aggregate results from multiple descendants (contributors, tags)
## Custom Non-ORM Relationships

## Resolver Options
For data sources that don't use ORM relationships (external APIs, caches, computed edges), declare custom `Relationship` on SQLModel entities with a hand-written async loader:

```python
result = await Resolver(
context={"user_id": 42}, # Pass global context
loader_params={}, # DataLoader extra parameters
).resolve(dtos)
from sqlmodel_nexus import Relationship as CustomRelationship

async def tags_loader(task_ids: list[int]) -> list[list[Tag]]:
async with get_session() as session:
stmt = select(Tag, TaskTag.task_id).join(TaskTag).where(TaskTag.task_id.in_(task_ids))
rows = (await session.exec(stmt)).all()
return build_list(rows, task_ids, lambda r: r[1], lambda r: r[0])

class Task(SQLModel, table=True):
__relationships__ = [
CustomRelationship(fk="id", target=list[Tag], name="tags", loader=tags_loader)
]
```

## Loader Dependency Name Rule
DTO auto-loads custom relationships the same way as ORM ones:

```python
class TaskDTO(DefineSubset):
__subset__ = (Task, ("id", "title"))
tags: list[TagDTO] = [] # Auto-loaded from the custom relationship
```

`Loader('author')` requires a relationship named `author` in ErManager. When using implicit auto-loading, you typically don't need to write Loaders manually.
> ⬆ This is **Level 5** in the demo: [`TaskWithTags`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/core_api/dtos.py#L140).

## Next Steps
## Execution Order (Full)

- [Custom Relationships](./custom_relationship.md) — Non-ORM relationship declarations
- [MCP Service](../advanced/mcp_service.md) — Expose APIs to AI agents
1. Execute all `resolve_*` methods on current node (load relationship data)
2. Traverse existing object/relationship fields recursively (depth-first)
3. Execute all `post_*` methods on current node (compute derived fields)
4. Collect SendTo values upward to ancestor Collectors
4 changes: 4 additions & 0 deletions docs/guide/graphql_mode.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,10 @@

From SQLModel entities to a complete GraphQL API — SDL auto-generation, automatic relationship resolution, and DataLoader batch loading.

> **Prerequisites**: You should be familiar with [SQLModel](https://sqlmodel.tiangolo.com/) entity definitions (SQLModel + Field + Relationship).
>
> **Live demo**: See [`demo/app.py`](https://github.com/allmonday/sqlmodel-nexus/blob/master/demo/app.py) for a working GraphQL server (User/Post/Comment).

## GraphQLHandler Configuration

```python
Expand Down
Loading
Loading