🤖 AI Agent Quick Reference Guide¶

# Commands (Write operations)
@dataclass
class CreateOrderCommand(Command[Order]):
    customer_id: str
    items: List[OrderItemDto]

class CreateOrderHandler(CommandHandler[CreateOrderCommand, Order]):
    async def handle_async(self, command: CreateOrderCommand) -> Order:
        # Business logic here
        return order

# Queries (Read operations)
@dataclass
class GetOrderQuery(Query[Optional[Order]]):
    order_id: str

class GetOrderHandler(QueryHandler[GetOrderQuery, Optional[Order]]):
    async def handle_async(self, query: GetOrderQuery) -> Optional[Order]:
        return await self.repository.get_by_id_async(query.order_id)

from neuroglia.mvc import ControllerBase
from classy_fastapi.decorators import get, post

class OrdersController(ControllerBase):
    @post("/", response_model=OrderDto, status_code=201)
    async def create_order(self, dto: CreateOrderDto) -> OrderDto:
        command = self.mapper.map(dto, CreateOrderCommand)
        order = await self.mediator.execute_async(command)
        return self.mapper.map(order, OrderDto)

    @get("/{order_id}", response_model=OrderDto)
    async def get_order(self, order_id: str) -> OrderDto:
        query = GetOrderQuery(order_id=order_id)
        order = await self.mediator.execute_async(query)
        return self.mapper.map(order, OrderDto)

# Abstract repository
class OrderRepository(Repository[Order, str]):
    async def get_by_customer_async(self, customer_id: str) -> List[Order]:
        pass

# MongoDB implementation
class MongoOrderRepository(MongoRepository[Order, str]):
    async def get_by_customer_async(self, customer_id: str) -> List[Order]:
        cursor = self.collection.find({"customer_id": customer_id})
        return [self._to_entity(doc) async for doc in cursor]

from neuroglia.hosting.web import WebApplicationBuilder
from neuroglia.mediation import Mediator
from neuroglia.mapping import Mapper

def create_app():
    builder = WebApplicationBuilder()

    # Configure core services
    Mediator.configure(builder, ["application.commands", "application.queries"])
    Mapper.configure(builder, ["application.mapping", "api.dtos", "domain.entities"])

    # Register custom services
    builder.services.add_scoped(OrderRepository, MongoOrderRepository)

    # Add SubApp with controllers
    builder.add_sub_app(
        SubAppConfig(
            path="/api",
            name="api",
            controllers=["api.controllers"]
        )
    )

    app = builder.build()
    return app

from neuroglia.observability import configure_opentelemetry, get_tracer, trace_async

# Configure OpenTelemetry (in main.py)
configure_opentelemetry(
    service_name="mario-pizzeria",
    service_version="1.0.0",
    otlp_endpoint="http://localhost:4317"
)

# Use in handlers
tracer = get_tracer(__name__)

class PlaceOrderHandler(CommandHandler):
    @trace_async(name="place_order")
    async def handle_async(self, command: PlaceOrderCommand):
        with tracer.start_as_current_span("validate_order"):
            # Business logic with automatic tracing
            pass

from fastapi import Depends
from fastapi.security import HTTPBearer

security = HTTPBearer()

class OrdersController(ControllerBase):
    @post("/", response_model=OrderDto)
    async def create_order(
        self,
        dto: CreateOrderDto,
        credentials: HTTPAuthorizationCredentials = Depends(security)
    ) -> OrderDto:
        # Extract user info from JWT
        user_info = self._decode_jwt(credentials.credentials)

        # Pass to handler for RBAC check
        command = CreateOrderCommand(
            customer_id=dto.customer_id,
            items=dto.items,
            user_info=user_info  # Handler checks roles/permissions
        )
        result = await self.mediator.execute_async(command)
        return self.process(result)

# RBAC in handler (application layer)
class CreateOrderHandler(CommandHandler):
    async def handle_async(self, command: CreateOrderCommand):
        # Authorization check based on roles
        if "customer" not in command.user_info.get("roles", []):
            return self.forbidden("Insufficient permissions")

        # Business logic
        order = Order(command.customer_id, command.items)
        await self.repository.save_async(order)
        return self.created(order)

# ✅ Use constructor injection
class OrderService:
    def __init__(self, repository: OrderRepository, event_bus: EventBus):
        self.repository = repository
        self.event_bus = event_bus

# ✅ Separate commands and queries
class PlaceOrderCommand(Command[Order]): pass
class GetOrderQuery(Query[Optional[Order]]): pass

# ✅ Use domain events
class Order(Entity):
    def place_order(self):
        # Business logic
        self.raise_event(OrderPlacedEvent(order_id=self.id))

# ✅ Type hints everywhere
async def handle_async(self, command: PlaceOrderCommand) -> Order:
    return order

# ❌ Direct database access in controllers
class OrderController:
    def create_order(self):
        # Don't access database directly
        connection.execute("INSERT INTO...")

# ❌ Mixing concerns
class OrderHandler:
    def handle(self, command):
        # Don't mix business logic with infrastructure
        send_email()  # Infrastructure concern

# ❌ Missing type hints
def process_order(order):  # What type is order?
    return result  # What type is result?

# Minimal Neuroglia application setup
from neuroglia.hosting.web import WebApplicationBuilder
from neuroglia.mediation import Mediator
from neuroglia.mapping import Mapper

def create_app():
    builder = WebApplicationBuilder()

    # Configure essential services
    Mediator.configure(builder, ["application.commands", "application.queries"])
    Mapper.configure(builder, ["application.mapping", "api.dtos"])

    # Add SubApp with controllers
    builder.add_sub_app(
        SubAppConfig(
            path="/api",
            name="api",
            controllers=["api.controllers"]
        )
    )

    # Build application
    app = builder.build()
    return app

if __name__ == "__main__":
    app = create_app()
    app.run()