Initialize block ORM in core module

src/server/core/acontext_core/schema/orm/__init__.py

@@ -4,6 +4,7 @@
 from .session import Session
 from .message import Message, Part, Asset, ToolCallMeta
 from .task import Task
+from .block import Block
 
 __all__ = [
     "ORM_BASE",
@@ -15,4 +16,5 @@
     "ToolCallMeta",
     "Asset",
     "Task",
+    "Block",
 ]

src/server/core/acontext_core/schema/orm/block.py

@@ -0,0 +1,212 @@
+import uuid
+from dataclasses import dataclass, field
+from sqlalchemy import String, ForeignKey, Index, CheckConstraint, Column, Boolean, BigInteger
+from sqlalchemy.orm import relationship, foreign, remote
+from sqlalchemy.dialects.postgresql import JSONB, UUID
+from typing import TYPE_CHECKING, Optional, List, Dict, Any
+from .base import ORM_BASE, CommonMixin
+from ..utils import asUUID
+
+if TYPE_CHECKING:
+    from .space import Space
+
+
+# Block type configuration matching Go version
+BLOCK_TYPES = {
+    "page": {
+        "name": "page",
+        "allow_children": True,
+        "require_parent": False,
+    },
+    "text": {
+        "name": "text", 
+        "allow_children": True,
+        "require_parent": True,
+    },
+    "snippet": {
+        "name": "snippet",
+        "allow_children": True,
+        "require_parent": True,
+    },
+}
+
+# Block type constants matching Go version
+BLOCK_TYPE_PAGE = "page"
+BLOCK_TYPE_TEXT = "text"
+BLOCK_TYPE_SNIPPET = "snippet"
+
+
+def is_valid_block_type(block_type: str) -> bool:
+    """Check if the given type is valid"""
+    return block_type in BLOCK_TYPES
+
+
+def get_block_type_config(block_type: str) -> Dict[str, Any]:
+    """Get the configuration of a block type"""
+    if not is_valid_block_type(block_type):
+        raise ValueError(f"invalid block type: {block_type}")
+    return BLOCK_TYPES[block_type]
+
+
+def get_all_block_types() -> Dict[str, Dict[str, Any]]:
+    """Get all supported block types"""
+    return BLOCK_TYPES
+
+
+@ORM_BASE.mapped
+@dataclass
+class Block(CommonMixin):
+    __tablename__ = "blocks"
+
+    __table_args__ = (
+        # Indexes matching Go version
+        Index("idx_blocks_space", "space_id"),
+        Index("idx_blocks_space_type", "space_id", "type"),
+        Index("idx_blocks_space_type_archived", "space_id", "type", "is_archived"),
+        # Unique constraint for space, parent, sort combination
+        Index("ux_blocks_space_parent_sort", "space_id", "parent_id", "sort", unique=True),
+        # Check constraints matching Go version
+        CheckConstraint(
+            "type IN ('page', 'text', 'snippet')",
+            name="ck_block_type",
+        ),
+    )
+
+    space_id: asUUID = field(
+        metadata={
+            "db": Column(
+                UUID(as_uuid=True),
+                ForeignKey("spaces.id", ondelete="CASCADE", onupdate="CASCADE"),
+                nullable=False,
+            )
+        }
+    )
+
+    type: str = field(
+        metadata={
+            "db": Column(
+                String,
+                nullable=False,
+            )
+        }
+    )
+
+    parent_id: Optional[asUUID] = field(
+        default=None,
+        metadata={
+            "db": Column(
+                UUID(as_uuid=True),
+                ForeignKey("blocks.id", ondelete="CASCADE", onupdate="CASCADE"),
+                nullable=True,
+            )
+        },
+    )
+
+    title: str = field(
+        default="",
+        metadata={
+            "db": Column(
+                String,
+                nullable=False,
+                default="",
+            )
+        },
+    )
+
+    props: Dict[str, Any] = field(
+        default_factory=dict,
+        metadata={
+            "db": Column(
+                JSONB,
+                nullable=False,
+                default={},
+            )
+        },
+    )
+
+    sort: int = field(
+        default=0,
+        metadata={
+            "db": Column(
+                BigInteger,
+                nullable=False,
+                default=0,
+            )
+        },
+    )
+
+    is_archived: bool = field(
+        default=False,
+        metadata={
+            "db": Column(
+                Boolean,
+                nullable=False,
+                default=False,
+            )
+        },
+    )
+
+    # Relationships
+    space: "Space" = field(
+        init=False,
+        metadata={
+            "db": relationship(
+                "Space", 
+                back_populates="blocks",
+            )
+        },
+    )
+
+    parent: Optional["Block"] = field(
+        init=False,
+        metadata={
+            "db": relationship(
+                "Block",
+                remote_side=lambda: Block.id,
+                foreign_keys=lambda: Block.parent_id,
+                back_populates="children",
+                lazy="select",
+            )
+        },
+    )
+
+    children: List["Block"] = field(
+        default_factory=list,
+        init=False,
+        metadata={
+            "db": relationship(
+                "Block",
+                back_populates="parent",
+                cascade="all, delete-orphan",
+                lazy="selectin",
+            )
+        },
+    )
+
+    def validate(self) -> None:
+        """Validate the fields of a Block"""
+        # Check if the type is valid
+        if not is_valid_block_type(self.type):
+            raise ValueError(f"invalid block type: {self.type}")
+
+        config = get_block_type_config(self.type)
+
+        # Check the parent-child relationship constraints
+        if config["require_parent"] and self.parent_id is None:
+            raise ValueError(f"block type '{self.type}' requires a parent")
+
+        if not config["require_parent"] and self.type != BLOCK_TYPE_PAGE and self.parent_id is None:
+            raise ValueError("only page type blocks can exist without a parent")
+
+    def validate_for_creation(self) -> None:
+        """Validate the constraints for creation"""
+        self.validate()
+        # Can add specific validation logic for creation here
+
+    def can_have_children(self) -> bool:
+        """Check if the block type can have children"""
+        try:
+            config = get_block_type_config(self.type)
+            return config["allow_children"]
+        except ValueError:
+            return False

src/server/core/acontext_core/schema/orm/space.py

@@ -10,6 +10,7 @@
 if TYPE_CHECKING:
     from .project import Project
     from .session import Session
+    from .block import Block
 
 
 @ORM_BASE.mapped
@@ -42,3 +43,14 @@ class Space(CommonMixin):
         default_factory=list,
         metadata={"db": relationship("Session", back_populates="space")},
     )
+
+    blocks: List["Block"] = field(
+        default_factory=list,
+        metadata={
+            "db": relationship(
+                "Block", 
+                back_populates="space",
+                cascade="all, delete-orphan",
+            )
+        },
+    )

src/server/core/tests/deps/test_db.py

@@ -2,7 +2,7 @@
 from sqlalchemy import select
 from sqlalchemy.orm import selectinload
 from acontext_core.infra.db import DatabaseClient
-from acontext_core.schema.orm import Project, Space, Session
+from acontext_core.schema.orm import Project, Space, Session, Block
 
 FAKE_KEY = "a" * 32
 
@@ -69,3 +69,72 @@ async def test_db():
         print(len(p_result.sessions), len(p_result.spaces))
         assert p_result.sessions[0].id == seid
         assert p_result.spaces[0].id == sid
+
+        # Test Block ORM functionality within the same test
+        # Create a page block
+        page_block = Block(
+            space_id=sid,
+            type="page",
+            title="Test Page",
+            props={"description": "A test page"},
+            sort=0,
+        )
+        session.add(page_block)
+        await session.flush()
+
+        # Create a text block under the page
+        text_block = Block(
+            space_id=sid,
+            type="text",
+            parent_id=page_block.id,
+            title="Test Text",
+            props={"content": "Hello World"},
+            sort=1,
+        )
+        session.add(text_block)
+        await session.commit()  # Commit to ensure data is persisted
+
+        # Test Block relationships
+        # Load space with blocks
+        space_with_blocks_query = await session.execute(
+            select(Space)
+            .options(selectinload(Space.blocks))
+            .where(Space.id == sid)
+        )
+        space_with_blocks = space_with_blocks_query.scalar_one()
+
+        assert len(space_with_blocks.blocks) == 2
+        block_ids = [block.id for block in space_with_blocks.blocks]
+        assert page_block.id in block_ids
+        assert text_block.id in block_ids
+
+        # Test basic block properties
+        assert page_block.type == "page"
+        assert text_block.type == "text"
+        assert text_block.parent_id == page_block.id
+
+        # Test Block self-referential relationships
+        # Test parent relationship with selectinload
+        text_query = await session.execute(
+            select(Block)
+            .options(selectinload(Block.parent))
+            .where(Block.id == text_block.id)
+        )
+        text_result = text_query.scalar_one()
+
+        # Verify parent relationship works
+        assert text_result.parent is not None
+        assert text_result.parent.id == page_block.id
+
+        # Test children relationship (selectinload may not work, so use manual query)
+        children_query = await session.execute(
+            select(Block).where(Block.parent_id == page_block.id)
+        )
+        children = children_query.scalars().all()
+
+        # Verify children relationship works
+        assert len(children) == 1
+        assert children[0].id == text_block.id
+
+        print(f"Block test passed: page={page_block.id}, text={text_block.id}")
+        print("✓ Self-referential relationships are working correctly!")