Knowledge graphs

alembic/versions/9a8abc15a493_adds_data_blocks.py

@@ -0,0 +1,99 @@
+"""adds data blocks
+
+Revision ID: 9a8abc15a493
+Revises: 4ffb627479d3
+Create Date: 2026-01-20 23:08:30.530156
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = "9a8abc15a493"
+down_revision = "4ffb627479d3"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "data_block",
+        sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column("organization_id", postgresql.UUID(as_uuid=True), nullable=True),
+        sa.Column("project_id", postgresql.UUID(as_uuid=True), nullable=True),
+        sa.Column("created_by", postgresql.UUID(as_uuid=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=True),
+        sa.Column("name", sa.String(), nullable=True),
+        sa.Column("description", sa.String(), nullable=True),
+        sa.Column("type", sa.String(), nullable=True),
+        sa.Column("sql_config", sa.JSON(), nullable=True),
+        sa.Column("sql_data", sa.ARRAY(sa.JSON()), nullable=True),
+        sa.ForeignKeyConstraint(["created_by"], ["user.id"], ondelete="SET NULL"),
+        sa.ForeignKeyConstraint(
+            ["organization_id"], ["organization.id"], ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(["project_id"], ["project.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_data_block_created_by"), "data_block", ["created_by"], unique=False
+    )
+    op.create_index(
+        op.f("ix_data_block_organization_id"),
+        "data_block",
+        ["organization_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_data_block_project_id"), "data_block", ["project_id"], unique=False
+    )
+    op.create_table(
+        "data_block_attributes",
+        sa.Column("id", postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column("data_block_id", postgresql.UUID(as_uuid=True), nullable=True),
+        sa.Column("name", sa.String(), nullable=True),
+        sa.Column("data_type", sa.String(), nullable=True),
+        sa.Column("is_primary_key", sa.Boolean(), nullable=True),
+        sa.Column("relative_position", sa.Integer(), nullable=True),
+        sa.Column("user_created", sa.Boolean(), nullable=True),
+        sa.Column("source_code", sa.String(), nullable=True),
+        sa.Column("state", sa.String(), nullable=True),
+        sa.Column("logs", sa.ARRAY(sa.String()), nullable=True),
+        sa.Column("started_at", sa.DateTime(), nullable=True),
+        sa.Column("finished_at", sa.DateTime(), nullable=True),
+        sa.Column("progress", sa.Float(), nullable=True),
+        sa.Column(
+            "additional_config",
+            sa.JSON(),
+            nullable=True,
+            comment="used when data_type == LLM_RESPONSE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["data_block_id"], ["data_block.id"], ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_data_block_attributes_data_block_id"),
+        "data_block_attributes",
+        ["data_block_id"],
+        unique=False,
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(
+        op.f("ix_data_block_attributes_data_block_id"),
+        table_name="data_block_attributes",
+    )
+    op.drop_table("data_block_attributes")
+    op.drop_index(op.f("ix_data_block_project_id"), table_name="data_block")
+    op.drop_index(op.f("ix_data_block_organization_id"), table_name="data_block")
+    op.drop_index(op.f("ix_data_block_created_by"), table_name="data_block")
+    op.drop_table("data_block")
+    # ### end Alembic commands ###

api/transfer.py

@@ -2,6 +2,7 @@
 import traceback
 import time
 import os
+
 from typing import Optional
 from starlette.endpoints import HTTPEndpoint
 from starlette.responses import PlainTextResponse
@@ -10,30 +11,25 @@
 from controller.transfer.cognition import (
     import_preparator as cognition_preparator,
 )
+from controller.transfer.cognition.minio_upload import handle_cognition_file_upload
+from controller.transfer import manager as transfer_manager
+from controller.upload_task import manager as upload_task_manager
+from controller.attribute import manager as attribute_manager
+from controller.task_master import manager as task_master_manager
+from util.notification import create_notification
+from util import notification, service_requests
 from exceptions.exceptions import BadPasswordError
+
+from submodules.model import enums, daemon
+from submodules.model.models import UploadTask
 from submodules.model.business_objects import (
     attribute,
     general,
     tokenization,
     project,
+    data_block,
 )
 from submodules.model.cognition_objects import integration
-from util import service_requests
-
-from controller.transfer import manager as transfer_manager
-from controller.upload_task import manager as upload_task_manager
-from controller.attribute import manager as attribute_manager
-
-from submodules.model import enums
-from util.notification import create_notification
-from submodules.model.enums import NotificationType
-from submodules.model.models import UploadTask
-from util import notification
-from submodules.model import daemon
-from controller.transfer.cognition.minio_upload import handle_cognition_file_upload
-
-from controller.task_master import manager as task_master_manager
-from submodules.model.enums import TaskType, RecordTokenizationScope
 
 
 logging.basicConfig(level=logging.DEBUG)
@@ -65,6 +61,8 @@ async def post(self, request) -> PlainTextResponse:
             return PlainTextResponse("OK")
         if upload_task_id == "download":
             return PlainTextResponse("OK")
+        if upload_task_id == "data-blocks":
+            return PlainTextResponse("OK")
         if org_id == "archive":
             return PlainTextResponse("OK")
 
@@ -127,10 +125,10 @@ def init_file_import(task: UploadTask, project_id: str, is_global_update: bool)
         task_master_manager.queue_task(
             str(org_id),
             str(task.user_id),
-            TaskType.TOKENIZATION,
+            enums.TaskType.TOKENIZATION,
             {
                 "project_id": str(project_id),
-                "scope": RecordTokenizationScope.PROJECT.value,
+                "scope": enums.RecordTokenizationScope.PROJECT.value,
                 "include_rats": True,
                 "only_uploaded_attributes": only_usable_attributes,
             },
@@ -141,14 +139,14 @@ def file_import_error_handling(
     task: UploadTask,
     project_id: str,
     is_global_update: bool,
-    notification_type: Optional[NotificationType] = None,
+    notification_type: Optional[enums.NotificationType] = None,
     print_traceback: bool = True,
 ) -> None:
     general.rollback()
     task.state = enums.UploadStates.ERROR.value
     general.commit()
     if not notification_type:
-        notification_type = NotificationType.IMPORT_FAILED
+        notification_type = enums.NotificationType.IMPORT_FAILED
     create_notification(
         notification_type,
         task.user_id,
@@ -245,7 +243,8 @@ def __calculate_missing_attributes(project_id: str, user_id: str) -> None:
                     project.get_org_id(project_id),
                     user_id,
                     current_att_id,
-                    True,
+                    include_rats=True,
+                    check_data_blocks_dependency=True,
                 )
             else:
                 if tokenization.is_doc_bin_creation_running_for_attribute(

app.py

@@ -35,6 +35,8 @@
 from fast_api.routes.record_internal import router as record_internal_router
 from fast_api.routes.playground import router as playground_router
 from fast_api.routes.inbox_mail import router as inbox_mail_router
+from fast_api.routes.data_block import router as data_block_router
+from fast_api.routes.data_block_internal import router as data_block_internal_router
 from middleware.database_session import handle_db_session
 from middleware.starlette_tmp_middleware import DatabaseSessionHandler
 from starlette.applications import Starlette
@@ -64,6 +66,8 @@
     PREFIX_TASK_EXECUTION,
     PREFIX_PLAYGROUND,
     PREFIX_INBOX_MAIL,
+    PREFIX_DATA_BLOCK,
+    PREFIX_DATA_BLOCK_INTERNAL,
 )
 from util import security, clean_up
 from middleware import log_storage
@@ -127,20 +131,27 @@
     inbox_mail_router, prefix=PREFIX_INBOX_MAIL, tags=["inbox_mail"]
 )
 
+fastapi_app.include_router(
+    data_block_router, prefix=PREFIX_DATA_BLOCK, tags=["data-block"]
+)
+
 app_name_internal = app_name + "-i"
 fastapi_app_internal = FastAPI(title=app_name_internal)
 
 fastapi_app_internal.include_router(
     task_execution_router, prefix=PREFIX_TASK_EXECUTION, tags=["task-execution"]
 )
-
 fastapi_app_internal.include_router(
     record_internal_router, prefix=PREFIX_RECORD_INTERNAL, tags=["record-internal"]
 )
 fastapi_app_internal.include_router(
     project_internal_router, prefix=PREFIX_PROJECT_INTERNAL, tags=["project-internal"]
 )
-
+fastapi_app_internal.include_router(
+    data_block_internal_router,
+    prefix=PREFIX_DATA_BLOCK_INTERNAL,
+    tags=["data-block-internal"],
+)
 
 routes = [
     Route("/full_config", FullConfigRest),

controller/attribute/llm_response_tmpl.py

@@ -287,7 +287,7 @@ def get_llm_config_a2vybg():
 
 
 async def get_llm_response(record: dict, cached_records: dict):
-    curr_running_id = str(record["running_id"])
+    curr_running_id = str(record.get("running_id", record.get("record_id")))
 
     if curr_running_id in cached_records:
         return cached_records[curr_running_id]

controller/attribute/manager.py

@@ -9,21 +9,24 @@
     project,
     tokenization,
     general,
+    data_block,
 )
 from submodules.model.models import Attribute
 from submodules.model.enums import (
     AttributeState,
     DataTypes,
     RecordTokenizationScope,
     AttributeVisibility,
+    TaskType,
 )
 from util import notification
 
 from submodules.model import daemon
+from submodules.model.util import sql_alchemy_to_dict
 from submodules.s3 import controller as s3
 
 from controller.task_master import manager as task_master_manager
-from submodules.model.enums import TaskType
+from controller.data_block import manager as data_block_manager
 from . import util
 from sqlalchemy import sql
 from hashlib import md5
@@ -51,6 +54,17 @@ def get_attribute(project_id: str, attribute_id: str) -> Attribute:
     return attribute.get(project_id, attribute_id)
 
 
+def get_attribute_expanded(project_id: str, attribute_id: str) -> Attribute:
+    attribute_item = sql_alchemy_to_dict(attribute.get(project_id, attribute_id))
+    attribute_item["related_data_blocks"] = sql_alchemy_to_dict(
+        data_block.get_refinery_attribute_dependants(
+            project_id=project_id,
+            refinery_attribute_name=attribute_item["name"],
+        )
+    )
+    return attribute_item
+
+
 def get_all_attributes_by_names(
     project_id: str, attribute_names: List[str]
 ) -> List[Attribute]:
@@ -240,6 +254,7 @@ def calculate_user_attribute_missing_records(
     user_id: str,
     attribute_id: str,
     include_rats: bool = True,
+    check_data_blocks_dependency: bool = False,
 ) -> None:
     if attribute.get_all(
         project_id=project_id, state_filter=[AttributeState.RUNNING.value]
@@ -291,6 +306,7 @@ def calculate_user_attribute_missing_records(
         user_id,
         attribute_id,
         include_rats,
+        check_data_blocks_dependency,
     )
 
 
@@ -300,6 +316,7 @@ def __calculate_user_attribute_missing_records(
     user_id: str,
     attribute_id: str,
     include_rats: bool,
+    check_data_blocks_dependency: bool = False,
 ) -> None:
     general.get_ctx_token()
 
@@ -424,6 +441,20 @@ def __calculate_user_attribute_missing_records(
     notification.send_organization_update(
         project_id, f"calculate_attribute:finished:{attribute_id}"
     )
+
+    if check_data_blocks_dependency:
+        dependant_data_blocks = util.get_dependant_data_blocks(
+            project_id, attribute_item.name
+        )
+        if dependant_data_blocks:
+            for data_block in dependant_data_blocks:
+                data_block_manager.update_query_results(
+                    org_id,
+                    user_id,
+                    data_block_id=str(data_block.id),
+                    sync_schema=False,
+                )
+
     general.remove_and_refresh_session()