💡 使用案例

📁 可用範例

  • quick_start.py - 基本 CRUD 操作

  • resource_crud.py - 完整功能演示

  • schema_upgrade.py - 數據遷移

  • backup.py - 備份與還原

� 完整程式碼

基本 CRUD 操作 (quick_start.py)

  1"""To start dev server, run
  2```
  3python -m fastapi dev quick_start.py
  4````
  5
  6To see run test http methods, run
  7```
  8python quick_start.py
  9```
 10
 11Model type choices are
 12"msgspec", "dataclass", "typeddict"
 13
 14"""
 15
 16import sys
 17from datetime import datetime, timedelta
 18
 19from fastapi import FastAPI
 20from fastapi.testclient import TestClient
 21
 22from autocrud import AutoCRUD
 23
 24if len(sys.argv) >= 2:
 25    mode = sys.argv[1]
 26else:
 27    mode = "msgspec"
 28
 29if mode not in (
 30    "msgspec",
 31    "dataclass",
 32    "typeddict",
 33):
 34    raise ValueError(f"Invalid mode: {mode}")
 35
 36
 37if mode == "msgspec":
 38    from msgspec import Struct
 39
 40    class TodoItem(Struct):
 41        title: str
 42        completed: bool
 43        due: datetime
 44
 45    class TodoList(Struct):
 46        items: list[TodoItem]
 47        notes: str
 48
 49elif mode == "dataclass":
 50    from dataclasses import dataclass
 51
 52    @dataclass
 53    class TodoItem:
 54        title: str
 55        completed: bool
 56        due: datetime
 57
 58    @dataclass
 59    class TodoList:
 60        items: list[TodoItem]
 61        notes: str
 62
 63
 64elif mode == "typeddict":
 65    from typing import TypedDict
 66
 67    class TodoItem(TypedDict):
 68        title: str
 69        completed: bool
 70        due: datetime
 71
 72    class TodoList(TypedDict):
 73        items: list[TodoItem]
 74        notes: str
 75
 76
 77crud = AutoCRUD()
 78crud.add_model(TodoItem)
 79crud.add_model(TodoList)
 80
 81app = FastAPI()
 82crud.apply(app)
 83
 84
 85def test():
 86    client = TestClient(app)
 87    resp = client.post(
 88        "/todo-list",
 89        json={"items": [], "notes": "my todo"},
 90    )
 91    print(resp.json())
 92    todo_list_id = resp.json()["resource_id"]
 93    resp = client.patch(
 94        f"/todo-list/{todo_list_id}",
 95        json=[
 96            {
 97                "op": "add",
 98                "path": "/items/-",
 99                "value": {
100                    "title": "Todo 1",
101                    "completed": False,
102                    "due": (datetime.now() + timedelta(hours=1)).isoformat(),
103                },
104            },
105        ],
106    )
107    print(resp.json())
108    resp = client.get(f"/todo-list/{todo_list_id}/data")
109    print(resp.json())
110    resp = client.patch(
111        f"/todo-list/{todo_list_id}",
112        json=[{"op": "replace", "path": "/items/0/completed", "value": True}],
113    )
114    resp = client.get(f"/todo-list/{todo_list_id}/data")
115    print(resp.json())
116
117
118if __name__ == "__main__":
119    test()

完整功能演示 (resource_crud.py)

  1"""To start dev server, run
  2```
  3python -m fastapi dev resource_crud.py
  4````
  5
  6To run test http methods, run
  7```
  8python resource_crud.py
  9```
 10
 11Model type choices are
 12"msgspec", "dataclass", "typeddict"
 13
 14"""
 15
 16import sys
 17
 18if len(sys.argv) >= 2:
 19    mode = sys.argv[1]
 20else:
 21    mode = "msgspec"
 22
 23if mode not in (
 24    "msgspec",
 25    "dataclass",
 26    "typeddict",
 27):
 28    raise ValueError(f"Invalid mode: {mode}")
 29
 30from datetime import datetime
 31
 32from fastapi import FastAPI
 33from fastapi.testclient import TestClient
 34
 35from autocrud import AutoCRUD
 36
 37if mode == "msgspec":
 38    from msgspec import Struct
 39
 40    class Product(Struct):
 41        name: str
 42        quantity: int
 43        price: int
 44        tags: list[str]
 45
 46elif mode == "dataclass":
 47    from dataclasses import dataclass
 48
 49    @dataclass
 50    class Product:
 51        name: str
 52        quantity: int
 53        price: int
 54        tags: list[str]
 55
 56elif mode == "typeddict":
 57    from typing import TypedDict
 58
 59    class Product(TypedDict):
 60        name: str
 61        quantity: int
 62        price: int
 63        tags: list[str]
 64
 65
 66crud = AutoCRUD()
 67crud.add_model(Product)
 68
 69app = FastAPI()
 70
 71crud.apply(app)
 72crud.openapi(app)
 73
 74
 75def test():
 76    try:
 77        import rich
 78
 79        console = rich.console.Console()
 80
 81        def print_section(s, *args, **kwargs):
 82            console.print(
 83                f"======{s}======",
 84                *args,
 85                style="red",
 86                **kwargs,
 87            )
 88
 89        print_json = console.print_json
 90    except ImportError:
 91        print_section = print
 92        print_json = print
 93
 94    client = TestClient(app)
 95
 96    print_section("Add 3 products")
 97    client.post(
 98        "/product",
 99        json={"name": "Apple", "quantity": 10, "price": 100, "tags": ["fruit", "food"]},
100    )
101    dt2 = datetime.now()
102    client.post(
103        "/product",
104        json={"name": "Banana", "quantity": 5, "price": 50, "tags": ["fruit", "food"]},
105    )
106    dt3 = datetime.now()
107    client.post(
108        "/product",
109        json={"name": "Cherry", "quantity": 2, "price": 25, "tags": ["fruit", "food"]},
110    )
111
112    print_section("Search for products created within a range")
113    resp = client.get(
114        "/product/full",
115        params={"created_time_end": dt3, "created_time_start": dt2},
116    )
117    print_json(resp.text)
118
119    print_section("Get meta of Banana")
120    banana = client.get(f"/product/{resp.json()[0]['meta']['resource_id']}/meta")
121    print_json(banana.text)
122    banana_resource_id = banana.json()["resource_id"]
123
124    print_section("Use patch to update the product")
125    resp = client.patch(
126        f"/product/{banana_resource_id}",
127        json=[
128            {"op": "replace", "path": "/quantity", "value": 20},
129            {"op": "add", "path": "/tags/-", "value": "snack"},
130            {"op": "move", "from": "/tags/0", "path": "/tags/-"},
131            {"op": "remove", "path": "/tags/0"},
132        ],
133    )
134    print_json(resp.text)
135
136    print_section("Use also use put to update the product")
137    resp = client.put(
138        f"/product/{banana_resource_id}",
139        json={"name": "Banana", "quantity": 5, "price": 250, "tags": ["fruit", "food"]},
140    )
141    print_json(resp.text)
142
143    new_banana = client.get(f"/product/{banana_resource_id}/data")
144    print_json(new_banana.text)
145
146    print_section("Switch back to the previous revision.")
147    client.post(
148        f"/product/{banana_resource_id}/switch/{banana.json()['current_revision_id']}",
149    )
150    banana = client.get(f"/product/{banana_resource_id}/data")
151    print_json(banana.text)
152
153    print_section("Delete the product")
154    client.delete(f"/product/{banana_resource_id}")
155    resp = client.get(f"/product/{banana_resource_id}/data")
156    print_json(resp.text)
157
158    print_section("Search for the deleted product")
159    resp = client.get("/product/meta", params={"is_deleted": True})
160    print_json(resp.text)
161    console.print(f"This is the same as {banana_resource_id=}")
162
163    print_section("Restore the deleted product")
164    resp = client.post(
165        f"/product/{banana_resource_id}/restore",
166        params={"is_deleted": True},
167    )
168    print_json(resp.text)
169    resp = client.get(f"/product/{banana_resource_id}/data")
170    console.print("After restoring, the product comes back")
171    print_json(resp.text)
172
173
174if __name__ == "__main__":
175    test()

數據遷移 (schema_upgrade.py)

  1"""To see run test http methods, run
  2```
  3python schema_upgrade.py
  4```
  5
  6Model type choices are
  7"msgspec", "dataclass", "typeddict"
  8
  9"""
 10
 11import shutil
 12import sys
 13from typing import IO
 14
 15
 16from autocrud.resource_manager.basic import Encoding, MsgspecSerializer
 17from autocrud.resource_manager.storage_factory import DiskStorageFactory
 18from autocrud.types import IMigration
 19
 20if len(sys.argv) >= 2:
 21    mode = sys.argv[1]
 22else:
 23    mode = "msgspec"
 24
 25if mode not in (
 26    "msgspec",
 27    "dataclass",
 28    "typeddict",
 29):
 30    raise ValueError(f"Invalid mode: {mode}")
 31
 32from fastapi import FastAPI
 33from fastapi.testclient import TestClient
 34
 35from autocrud import AutoCRUD
 36
 37
 38def get_after_user():
 39    if mode == "msgspec":
 40        from msgspec import Struct
 41
 42        class User(Struct):
 43            name: str
 44            age: int
 45
 46    elif mode == "dataclass":
 47        from dataclasses import dataclass
 48
 49        @dataclass
 50        class User:
 51            name: str
 52            age: int
 53
 54    elif mode == "typeddict":
 55        from typing import TypedDict
 56
 57        class User(TypedDict):
 58            name: str
 59            age: int
 60
 61    return User
 62
 63
 64def get_before_user():
 65    if mode == "msgspec":
 66        from msgspec import Struct
 67
 68        class User(Struct):
 69            name: str
 70            income: float
 71
 72    elif mode == "dataclass":
 73        from dataclasses import dataclass
 74
 75        @dataclass
 76        class User:
 77            name: str
 78            income: float
 79
 80    elif mode == "typeddict":
 81        from typing import TypedDict
 82
 83        class User(TypedDict):
 84            name: str
 85            income: float
 86
 87    return User
 88
 89
 90def apply(before_after):
 91    if before_after == "before":
 92        shutil.rmtree("_autocrud_test_resource_dir", ignore_errors=True)
 93    User = get_before_user() if before_after == "before" else get_after_user()
 94
 95    crud = AutoCRUD(storage_factory=DiskStorageFactory("_autocrud_test_resource_dir"))
 96
 97    class Migration(IMigration):
 98        @property
 99        def schema_version(self):
100            return "v1"
101
102        def migrate(self, data: IO[bytes], schema_version: str | None):
103            BeforeUser = get_before_user()
104            s = MsgspecSerializer(encoding=Encoding.json, resource_type=BeforeUser)
105            od = s.decode(data.read())
106            if mode == "typeddict":
107                newd = User(
108                    name=od["name"],
109                    age=-1,
110                )
111            else:
112                newd = User(
113                    name=od.name,
114                    age=-1,
115                )
116            return newd
117
118    crud.add_model(
119        User,
120        migration=None if before_after == "before" else Migration(),
121    )
122
123    app = FastAPI()
124    crud.apply(app)
125    return app
126
127
128def test_before():
129    app = apply("before")
130    client = TestClient(app)
131    resp = client.post(
132        "/user",
133        json={"name": "John", "income": 100},
134    )
135    resp.raise_for_status()
136    print(resp.json())
137    resource_id = resp.json()["resource_id"]
138    resp = client.get(f"/user/{resource_id}/data")
139
140
141def test_after():
142    app = apply("after")
143    client = TestClient(app)
144    resp = client.get(
145        "/user/full",
146    )
147    resp.raise_for_status()
148    print(resp.json())
149    resource_id = resp.json()[0]["revision_info"]["resource_id"]
150    resp = client.patch(
151        f"/user/{resource_id}",
152        json=[
153            {"op": "replace", "path": "/age", "value": 10},
154        ],
155    )
156    print(resp.json())
157    resp = client.get(
158        f"/user/{resource_id}/full",
159    )
160    print(resp.json())
161
162
163if __name__ == "__main__":
164    test_before()
165    test_after()

備份與還原 (backup.py)

  1"""To see run test http methods, run
  2```
  3python schema_upgrade.py
  4```
  5
  6Model type choices are
  7"msgspec", "dataclass", "typeddict"
  8
  9"""
 10
 11import shutil
 12import sys
 13
 14from autocrud.resource_manager.storage_factory import DiskStorageFactory
 15
 16if len(sys.argv) >= 2:
 17    mode = sys.argv[1]
 18else:
 19    mode = "msgspec"
 20
 21if mode not in (
 22    "msgspec",
 23    "dataclass",
 24    "typeddict",
 25):
 26    raise ValueError(f"Invalid mode: {mode}")
 27
 28from fastapi import FastAPI
 29from fastapi.testclient import TestClient
 30
 31from autocrud import AutoCRUD
 32
 33if mode == "msgspec":
 34    from msgspec import Struct
 35
 36    class User(Struct):
 37        name: str
 38        age: int
 39
 40elif mode == "dataclass":
 41    from dataclasses import dataclass
 42
 43    @dataclass
 44    class User:
 45        name: str
 46        age: int
 47
 48elif mode == "typeddict":
 49    from typing import TypedDict
 50
 51    class User(TypedDict):
 52        name: str
 53        age: int
 54
 55
 56def apply():
 57    shutil.rmtree("_autocrud_test_resource_dir", ignore_errors=True)
 58
 59    crud = AutoCRUD(storage_factory=DiskStorageFactory("_autocrud_test_resource_dir"))
 60    crud.add_model(User)
 61
 62    app = FastAPI()
 63    crud.apply(app)
 64    return app, crud
 65
 66
 67def test_before():
 68    app, crud = apply()
 69    client = TestClient(app)
 70    resp = client.post(
 71        "/user",
 72        json={"name": "John", "age": 42},
 73    )
 74    resp.raise_for_status()
 75    print(resp.json())
 76    resource_id = resp.json()["resource_id"]
 77    resp = client.get(f"/user/{resource_id}/data")
 78    with open("_autocrud.dump", "wb") as f:
 79        crud.dump(f)
 80
 81
 82def test_after():
 83    app, crud = apply()
 84    with open("_autocrud.dump", "rb") as f:
 85        crud.load(f)
 86    client = TestClient(app)
 87    resp = client.get(
 88        "/user/full",
 89    )
 90    resp.raise_for_status()
 91    print(resp.json())
 92    resource_id = resp.json()[0]["revision_info"]["resource_id"]
 93    resp = client.patch(
 94        f"/user/{resource_id}",
 95        json=[
 96            {"op": "replace", "path": "/age", "value": 10},
 97        ],
 98    )
 99    print(resp.json())
100    resp = client.get(
101        f"/user/{resource_id}/full",
102    )
103    print(resp.json())
104
105
106if __name__ == "__main__":
107    test_before()
108    test_after()

🔍 自動生成的端點

方法

路徑

功能

POST

/model

創建資源

GET

/model/{id}/data

獲取數據

GET

/model/{id}/meta

獲取元數據

GET

/model/{id}/full

獲取完整資源

PUT

/model/{id}

完整更新

PATCH

/model/{id}

JSON Patch 更新

DELETE

/model/{id}

軟刪除

GET

/model/data

列出所有數據

GET

/model/meta

列出所有元數據

GET

/model/full

列出完整資源

POST

/model/{id}/switch/{revision}

版本切換

POST

/model/{id}/restore

恢復已刪除

🚀 運行範例

# 基本範例
python examples/quick_start.py
python examples/resource_crud.py

# 數據遷移
python examples/schema_upgrade.py

# 備份還原
python examples/backup.py

# 不同數據類型
python examples/quick_start.py pydantic
python examples/resource_crud.py dataclass

# 開發服務器
python -m fastapi dev examples/quick_start.py