GitHub - prostomarkeloff/emergent: Python is all you need.

@schema_meta(http_crud("/users", Users))
@dataclass
class User:
    id: Annotated[int, Identity]
    name: str
    email: Annotated[str, Unique, MaxLen(255)]

uv add git+https://github.com/prostomarkeloff/emergent.git

@schema_meta(
    http_crud("/bounties", Board, ops=(LIST, GET, CREATE)),
    Methods(),
)
@dataclass
class Bounty:
    id: Annotated[int, Identity]
    title: str
    reward: Annotated[int, Min(0)]
    status: Annotated[str, MaxLen(20), Doc("Bounty status"),
                       cli.Help("Current status"), openapi.Description("Status field"),
                       sql.Index("idx_status")]

    @classmethod
    @post("/bounties/{bounty_id}/claim")
    async def claim(cls, db: ..., bounty_id: int, hunter: str) -> Result[Bounty, DomainError]: ...

from dataclasses import dataclass, replace
from typing import Annotated, Protocol, runtime_checkable

from emergent.wire.compile._core import Axes
from emergent.wire.compile._phase import CompilationPhase
from emergent.wire.compile import to_pydantic, to_json_schema
from emergent.wire.axis.schema import Identity, MaxLen, Min
from emergent.wire.axis.schema._universal import SchemaAxisCapability
from emergent.wire.axis._capability import OpenAPIContext, openapi_schema


# ── Your language: search index ──────────────────────────────

@dataclass(frozen=True, slots=True)
class IndexFieldCtx:
    field_name: str = ""
    field_type: type = object
    searchable: bool = False
    boost: float = 1.0
    facet: bool = False


@runtime_checkable
class IndexCompilable(Protocol):
    def compile_index(self, ctx: IndexFieldCtx) -> IndexFieldCtx: ...


INDEX_PHASE = CompilationPhase(
    IndexFieldCtx,
    IndexCompilable,
    lambda n, t: IndexFieldCtx(field_name=n, field_type=t),
)


# ── Capabilities: one atom, two languages ────────────────────

@dataclass(frozen=True, slots=True)
class Searchable(SchemaAxisCapability):
    boost: float = 1.0

    # YOUR language
    def compile_index(self, ctx: IndexFieldCtx) -> IndexFieldCtx:  
        return replace(ctx, searchable=True, boost=self.boost)

    # emergent's!
    def compile_openapi(self, ctx: OpenAPIContext) -> OpenAPIContext:
        return openapi_schema(
            ctx, **{"x-searchable": True, "x-search-boost": self.boost}
        )


@dataclass(frozen=True, slots=True)
class Facet(SchemaAxisCapability):
    def compile_index(self, ctx: IndexFieldCtx) -> IndexFieldCtx:
        return replace(ctx, facet=True)

    def compile_openapi(self, ctx: OpenAPIContext) -> OpenAPIContext:
        return openapi_schema(ctx, **{"x-facet": True})


# ── Your compiler: to_search_index — same shape as to_pydantic

@dataclass(frozen=True, slots=True)
class IndexField:
    name: str
    searchable: bool = False
    boost: float = 1.0
    facet: bool = False


@dataclass(frozen=True, slots=True)
class SearchIndex:
    entity: str
    fields: tuple[IndexField, ...]


INDEX_COMPILER = SchemaCompiler(phases=(INDEX_PHASE,))


def to_search_index(cls: type, axes: Axes) -> SearchIndex:
    ec = INDEX_COMPILER.compile(cls, axes)
    return SearchIndex(
        entity=cls.__name__,
        fields=tuple(
            IndexField(
                name=c.field_name,
                searchable=c.searchable,
                boost=c.boost,
                facet=c.facet,
            )
            for fc in ec
            if (c := fc[INDEX_PHASE]).searchable or c.facet
        ),
    )


# ── Entity ───────────────────────────────────────────────────

@dataclass
class Product:
    id: Annotated[int, Identity]
    name: Annotated[str, MaxLen(200), Searchable(boost=3.0)]
    description: Annotated[str, Searchable()]
    category: Annotated[str, Facet()]
    price: Annotated[float, Min(0), Facet()]


# ── Same axes, three compilers ───────────────────────────────

axes = Axes.default()

index  = to_search_index(Product, axes)   # YOUR compiler   → SearchIndex
model  = to_pydantic(Product, axes)       # emergent's       → Pydantic BaseModel
schema = to_json_schema(Product, axes)    # emergent's       → OpenAPI JSON Schema

index → SearchIndex(entity='Product', fields=(
    IndexField(name='name', searchable=True, boost=3.0),
    IndexField(name='description', searchable=True, boost=1.0),
    IndexField(name='category', facet=True),
    IndexField(name='price', facet=True),
))

model → <class 'Product'>  (Pydantic BaseModel with MaxLen, Min validation)

schema → {
    "name": {"type": "string", "maxLength": 200, "x-searchable": true, "x-search-boost": 3.0},
    "category": {"type": "string", "x-facet": true},
    "price": {"type": "number", "minimum": 0, "x-facet": true},
    ...
}

@G.node
class FetchPrice:
    @classmethod
    async def __compose__(cls, product: Product, db: Database) -> float:
        return await db.get_price(product.id)

@G.node
class FetchStock:
    @classmethod
    async def __compose__(cls, product: Product, db: Database) -> int:
        return await db.get_stock(product.id)

# FetchPrice and FetchStock have no dependency on each other → run in parallel.
# No asyncio.gather. No concurrency code. The type signatures ARE the graph.
result = await G.compose(BuildReport, product, db)

explain_schema(User)
# === User ===
#   [SchemaName('users')]
#
#   id (int):
#     [Identity]
#
#   email (str):
#     [Unique, MaxLen(255)]
#     cli: Help('Email address')
#     openapi: Description('User email'), Format('email')
#     sql: Index('idx_email')

explain_application(app)
# === Application (3 endpoints) ===
#
#   Endpoint #1 (2 exposures):
#     [POST /api/v1/players] RequestResponseCodec
#       request: RegisterRequest, response: RegisterResponse
#     [register (cli)] RequestResponseCodec
#       request: RegisterRequest, response: RegisterResponse

explain_full(User)  # 3-layer trace
# === User (full trace) ===
#
#   Schema: 3 fields (1 identity)
#     id (int) [Identity]
#     name (str)
#     email (str) [Unique]
#
#   Derivation: 1 pattern
#     Pattern #1: Dialect (11 steps), provider=Users
#       steps: InspectEntity → RequireIdentity → BindProvider → ...
#       ops: List, Get, Create, Update, Patch, Delete
#
#   Surface: 6 exposures across 1 endpoint
#     GET /api/users [ListUserRequest → ListUserResponse]
#     GET /api/users/{id} [GetUserRequest → GetUserResponse]
#     ...

from emergent.wire.verify import verify_raising

@dataclass
class Sensor:
    id: Annotated[int, Identity]
    name: Annotated[str, MinLen(50), MaxLen(10)]       # ERROR: MinLen > MaxLen
    temp: Annotated[float, Min(200), Max(125)]          # ERROR: Min > Max
    secret: Annotated[str, ReadOnly(), WriteOnly()]      # ERROR: inaccessible field

verify_raising(Sensor)  # VerificationError with all issues

# Level 1 — auto CRUD: one decorator, full API
@schema_meta(http_crud("/products", Store))
@dataclass
class Product:
    id: Annotated[int, Identity]
    name: str
    price: float

# Level 2 — CRUD + methods: derive CRUD, hand-write domain logic
@schema_meta(http_crud("/bounties", Board, ops=(LIST, GET, CREATE)), Methods())
@dataclass
class Bounty:
    id: Annotated[int, Identity]
    title: str
    reward: int

    @classmethod
    @post("/bounties/{bounty_id}/claim")
    async def claim(cls, db: ..., bounty_id: int, hunter: str) -> Result[Bounty, DomainError]: ...

# Level 3 — hand-written methods: every endpoint explicit, full control
@schema_meta(Methods())
@dataclass
class Auth:
    @classmethod
    @post("/login")
    async def login(cls, creds: Credentials) -> Result[Token, AuthError]: ...

# Level 4 — raw wire: one endpoint, three targets
endpoint(runner)
    .expose(HTTPRouteTrigger("POST", "/register"), rrc(Req, Resp))
    .expose(CLITrigger("register"), rrc(Req, Resp))
    .expose(TelegrindTrigger(Command("register")), rrc(Req, Resp))

@schema_meta(
    http_crud("/posts", Posts),
    WithoutDelete(),    # ← removes DELETE op across all targets
    Paginated(20),      # ← adds pagination to ops with Pageable effect
    Sorted(),           # ← adds sorting to LIST
    Filtered("author"), # ← adds exact-match filter
)
@dataclass
class Post:
    id: Annotated[int, Identity]
    title: str
    body: str
    author: str

Define a context. Define a protocol. Call fold.

You just created a compilation language.