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.
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ì.
app = ()
- 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.
import sqlalchemy as safrom fluvius.data import SqlaDataSchema, SqlaDriver, DataAccessManagerfrom sqlalchemy.dialects import postgresql as pgfrom fluvius.data import UUID_GENRclass 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.
- Lời gọiHTTP · MCPMột request đến qua FastAPI hoặc một lời gọi công cụ MCP.
- Commandcreate-userCreateUserCmd kiểm tra payload theo Data model của nó.
- AggregateUserAggregate.create_userDecorator @action đặt tên sự kiện và áp dụng thay đổi.
- Nhật ký sự kiệnuser-createdSQLDomainLogStore ghi thêm sự kiện vào nhật ký miền.
- QueryUserQueryRead model phục vụ trạng thái mới qua các trường có kiểu.
- _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.
Truy cập
Vận hành
Lịch sử
Nền và thời gian thực
Tác nhân
Cài phần lõi. Thêm gói khi bạn cần.
Mô hình
Phục vụ
Nhóm phụ thuộc tùy chọn
python -m pip install "fluvius[workq]"Broker Taskiq cho Redis, NATS và PostgreSQLpython -m pip install "fluvius[transport]"Client vận chuyển MQTT và NATSpython -m pip install "fluvius[hatchet]"Tích hợp quy trình Hatchet
Năm phút để có một API đang chạy
- 1Cài đặt
python -m pip install fluvius - 2Ghép ứng dụng
app = create_app(pkginfo) | configure_routes() - 3Chạy
uvicorn 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.