Chuyển tới nội dung chính

Mô tả miền.
Phục vụ qua HTTP và MCP.

Fluvius là framework Python 3.12+ cho backend theo hướng miền. Bạn viết command, aggregate và query. Fluvius ghi mọi thay đổi vào nhật ký sự kiện và phục vụ miền qua FastAPI và MCP.

python -m pip install fluvius
Đọc phần bắt đầu nhanhXem tài liệu

Python 3.12+FastAPI · MCP

nhật ký sự kiện · user-profilechỉ ghi thêm
  1. 001030user-updatedupdate-useruser:2d9e0a47
  2. 001029user-invalidatedinvalidate-useruser:7f3a91c2
  3. 001028user-createdcreate-useruser:2d9e0a47
  4. 001027user-updatedupdate-useruser:c04e5b18
  5. 001026user-createdcreate-useruser:c04e5b18
  6. 001025user-updatedupdate-useruser:7f3a91c2
  7. 001024user-createdcreate-useruser:7f3a91c2

fluvius.fastapi · fluvius.mcp

Ghép ứng dụng, từng pipe một

Configurator nhận ứng dụng và trả lại nó, nên thứ tự là rõ ràng. Tắt một dòng để thấy ứng dụng ngừng phục vụ gì.

Đọc tài liệu
app/main.py
app = ()
Ứng dụng phục vụ
  • HTTP/api/docsGiao diện OpenAPI
  • HTTP/healthKiểm tra sức khỏe và readiness của framework
  • HTTP/api/authĐăng nhập Keycloak / OpenID Connect, phiên, CSRF
  • HTTP/api/user-profile/create-user:post/userEndpoint của command
  • HTTP/api/user-profile/update-user:exec/user/{id}Endpoint của command
  • HTTP/api/user-profile/invalidate-user:exec/user/{id}Endpoint của command

6 bề mặt HTTP · 0 endpoint MCP

fluvius.domain · fluvius.data · fluvius.query

Xây một miền trong sáu tệp

Ví dụ user-profile trong tài liệu, theo thứ tự bạn sẽ viết.

Đọc tài liệu
user_domain/model.py
import sqlalchemy as sa
from fluvius.data import SqlaDataSchema, SqlaDriver, DataAccessManager
from sqlalchemy.dialects import postgresql as pg
from fluvius.data import UUID_GENR
class UserConnector(SqlaDriver):
__db_dsn__ = "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres"
class User(UserConnector.__data_schema_base__):
__tablename__ = "user"
_id = sa.Column(pg.UUID, primary_key=True, default=UUID_GENR)
name = sa.Column(sa.String)
_created = sa.Column(sa.DateTime(timezone=True))
_updated = sa.Column(sa.DateTime(timezone=True))
_etag = sa.Column(sa.String)
_deleted = sa.Column(sa.DateTime(timezone=True))
_creator = sa.Column(pg.UUID)
_updater = sa.Column(pg.UUID)
_realm = sa.Column(sa.String)
class FluviusAccessManager(DataAccessManager):
__connector__ = UserConnector
__automodel__ = True

fluvius.domain

Chuyện gì xảy ra với một command

Một lời gọi trở thành command, aggregate áp dụng nó, và kết quả được ghi thêm vào nhật ký sự kiện. Kéo nhật ký để replay một user.

Đọc tài liệu
  1. Lời gọiHTTP · MCPMột request đến qua FastAPI hoặc một lời gọi công cụ MCP.
  2. Commandcreate-userCreateUserCmd kiểm tra payload theo Data model của nó.
  3. AggregateUserAggregate.create_userDecorator @action đặt tên sự kiện và áp dụng thay đổi.
  4. Nhật ký sự kiệnuser-createdSQLDomainLogStore ghi thêm sự kiện vào nhật ký miền.
  5. QueryUserQueryRead model phục vụ trạng thái mới qua các trường có kiểu.
Phát lại user:7f3a91c2seq 001036 · 4 / 4 sự kiện
_id
7f3a91c2
name
"Ada King"
_created
09:41:07
_updated
10:15:40
_deleted
10:15:40

Nhật ký sự kiện là lịch sử. Replay nó để xem trạng thái tại một thời điểm.

Có sẵn trong bộ khung

Những phần hầu hết backend phải tự dựng, đã nối vào cùng một miền.

Tham chiếu API công khai

Truy cập

Đăng nhập
Route Keycloak / OpenID Connect, phiên, bảo vệ CSRF
fluvius.fastapi
Phân quyền
roles_required trên command; policy Casbin khi bật policy_required
fluvius.casbin

Vận hành

Lỗi
Phản hồi application/problem+json và request ID
fluvius.fastapi
Sức khỏe
/health và trạng thái readiness
fluvius.fastapi

Lịch sử

Nhật ký sự kiện
SQLDomainLogStore ghi các sự kiện mà mỗi command phát ra
fluvius.domain
Lưu trữ
SQLAlchemy; PostgreSQL khi chạy thật, SQLite cho kiểm thử tập trung
fluvius.data

Nền và thời gian thực

Tác vụ và cron
Taskiq trên broker PostgreSQL, Redis hoặc NATS
fluvius.workq
Vận chuyển
NATS và MQTT, cầu nối thời gian thực, codec JSON hoặc CBOR
fluvius.transport

Tác nhân

MCP
Công cụ Streamable HTTP; JWT ngắn hạn, mặc định 900 giây
fluvius.mcp

Năm phút để có một API đang chạy

  1. 1Cài đặtpython -m pip install fluvius
  2. 2Ghép ứng dụngapp = create_app(pkginfo) | configure_routes()
  3. 3Chạyuvicorn myapp.main:app --reloadMở /docs để xem OpenAPI và /health để xem sức khỏe framework.

Bắt đầu từ phần bắt đầu nhanh.