Справочник по API Graph Builder (предварительная версия)

Класс sentinel_graph предоставляет способ взаимодействия с графом Microsoft Sentinel, позволяя определять схему графа, преобразовывать данные из озера данных Microsoft Sentinel в узлы и ребра, публиковать граф, запрашивать граф и запускать расширенные алгоритмы графа. Этот класс предназначен для работы с сеансами Spark в записных книжках Jupyter, работающих на Microsoft Sentinel вычислений Spark.

GraphSpecBuilder

Класс GraphSpecBuilder предоставляет построитель fluent для создания спецификаций графов с конвейерами данных и интеграцией схем.

# Deprecated — emits DeprecationWarning
from sentinel_graph.builders.graph_builder import GraphBuilder

# Recommended
from sentinel_graph import GraphSpecBuilder

constructor

GraphSpecBuilder(context: ExecutionContext)

Параметры:

• context (ExecutionContext): контекст выполнения, содержащий сеанс и конфигурацию Spark.

Поднимает:

• ValueError: если контекст имеет значение None или не удается определить имя графа.

Статические методы

start

GraphSpecBuilder.start(context: Optional[ExecutionContext] = None) -> GraphSpecBuilder

Определите новый построитель графов с бегучим потоком.

Параметры:

• context (ExecutionContext, необязательно): экземпляр ExecutionContext. Если нет, использует контекст по умолчанию.

Возвращает:

• GraphSpecBuilder: новый экземпляр построителя

Пример:

builder = GraphSpecBuilder.start(context=context)

Методы экземпляра

add_node

def add_node(alias: str) -> NodeBuilderInitial

Начните создание определения узла.

Параметры:

• alias (str): уникальный идентификатор для этого узла в графе

Возвращает:

• NodeBuilderInitial: построитель узлов в начальном состоянии

Пример:

builder.add_node("user")

add_edge

def add_edge(alias: str) -> EdgeBuilderInitial

Начните создание определения границы.

Параметры:

• alias (str): идентификатор для этого края в графе (может быть общим для нескольких ребер)

Возвращает:

• EdgeBuilderInitial: построитель Edge в начальном состоянии

Пример:

builder.add_edge("accessed")

done

def done() -> GraphSpec

Завершите спецификацию графа и верните экземпляр GraphSpec.

Возвращает:

• GraphSpec: полная спецификация графа с конвейером данных и схемой

Поднимает:

• ValueError: если граф не имеет узлов или ребер, или если проверка завершается ошибкой.

Пример:

graph_spec = builder.done()

GraphSpec

Спецификация графа с конвейером данных, схемой и возможностями отображения.

constructor

GraphSpec(
    name: str,
    context: ExecutionContext,
    graph_schema: GraphSchema,
    etl_pipeline: Optional[ETLPipeline] = None
)

Параметры:

• name (str): имя графа
• context (ExecutionContext): контекст выполнения
• graph_schema (GraphSchema): определение схемы графа
• etl_pipeline (ETLPipeline, необязательно): конвейер данных для подготовки графа

Свойства

nodes

def nodes() -> DataFrame

Получение узлов DataFrame (отложенный, кэшированный). Автоматически определяет источник из конвейера данных или таблицы озера.

Возвращает:

• DataFrame: кадр данных Spark, содержащий все узлы

Поднимает:

• ValueError: если контекст отсутствует или не удается загрузить кадры данных.

edges

def edges() -> DataFrame

Получение ребер DataFrame (отложенный, кэшированный). Автоматически определяет источник из конвейера данных или таблицы озера.

Возвращает:

• DataFrame: кадр данных Spark, содержащий все ребра.

Поднимает:

• ValueError: если контекст отсутствует или не удается загрузить кадры данных.

Методы

build_graph_with_data

def build_graph_with_data() -> Dict[str, Any]

Выполните конвейер данных и опубликуйте граф. Внутренний вызов , Graph.build(self)содержит возвращенный Graphи возвращает словарь, совместимый с обратной совместимостью.

Возвращает:

• Dict[str, Any]: словарь, содержащий:
  • etl_result: результаты подготовки данных
  • api_result: публикация результатов (в случае успешного выполнения)
  • api_error: строка ошибки (если публикация завершилась сбоем).
  • instance_name: имя экземпляра Graph
  • status: "published" или "prepared"

Пример:

graph = Graph.build(spec)
print(f"Status: {graph.build_status.status}")

get_schema

def get_schema() -> GraphSchema

Получение схемы графа.

Возвращает:

• GraphSchema: определение схемы графа

get_pipeline

def get_pipeline() -> Optional[ETLPipeline]

Получение конвейера данных (нет для существующих графов).

Возвращает:

• ETLPipeline или None: конвейер данных, если он доступен

to_graphframe

def to_graphframe(column_mapping: Optional[Dict[str, str]] = None) -> GraphFrame

Преобразование всего графа в GraphFrame для выполнения алгоритмов графа. Работает только с локальными данными (из конвейера данных или таблицы озера).

Параметры:

• column_mapping (Dict[str, str], необязательно): настраиваемое сопоставление столбцов с ключами:
  • "id": имя столбца идентификатора вершины
  • "source_id": имя столбца исходного идентификатора edge
  • "target_id": имя столбца идентификатора целевого объекта edge

Возвращает:

• GraphFrame: объект GraphFrame со всеми вершинами и ребрами

Поднимает:

• ValueError: если ExecutionContext недоступен

Пример:

gf = graph_spec.to_graphframe()
pagerank = gf.pageRank(resetProbability=0.15, maxIter=10)

show

def show(limit: int = 100, viz_format: str = "visual") -> None

Отображение данных графа в различных форматах.

Параметры:

• limit (int, default=100): максимальное число узлов и ребер для отображения
• viz_format (str, default="visual"): формат вывода
  • "table": полные таблицы dataframe (все столбцы)
  • "visual": интерактивная визуализация графа
  • "all": отображение всех форматов

Поднимает:

• ValueError: если формат не является одним из поддерживаемых значений.

Пример:

graph_spec.show(limit=50, viz_format="table")

show_schema

def show_schema() -> None

Отображение схемы графа в виде интерактивной визуализации графа.

Пример:

spec.show_schema()

Microsoft Graph

Запрашиваемый экземпляр графа. Создано с помощью Graph.get() (существующего графаGraph.build()) или GraphSpec (из ).

constructor

Graph(
    name: str,
    context: ExecutionContext,
    spec: Optional[GraphSpec] = None,
    build_status: Optional[BuildStatus] = None,
)

Параметры:

• name (str): имя графа
• context (ExecutionContext): контекст выполнения
• spec (GraphSpec, необязательно): спецификация присоединенного графа (задается параметром Graph.build())
• build_status(BuildStatus, необязательно): метаданные результата сборки (задаются )Graph.build()

Поднимает:

• ValueError: если Параметр ExecutionContext имеет значение None.

Статические методы

get

Graph.get(name: str, context: Optional[ExecutionContext] = None) -> Graph

Получение экземпляра графа из существующего графа. Возвращенный Graph имеет spec=None и build_status=None.

Параметры:

• name (str): имя экземпляра Graph
• context (ExecutionContext, необязательно): контекст выполнения (по умолчанию — ExecutionContext.default())

Возвращает:

• Graph: экземпляр Graph

Поднимает:

• ValueError: если имя графа пустое или экземпляр графа не существует.

Пример:

graph = Graph.get("my_graph", context=context)
graph.query("MATCH (n) RETURN n")

prepare

Graph.prepare(spec: GraphSpec) -> Graph

Запустите этап подготовки данных для GraphSpec публикации. Затем используйте publish() , чтобы зарегистрировать граф и сделать его запрашиваемым.

Параметры:

• spec (GraphSpec): спецификация графа для подготовки

Возвращает:

• Graph: экземпляр Graph с spec присоединенным и build_status.status == "prepared"

Поднимает:

• ValueError: если спецификация не имеет конвейера данных или контекста выполнения.
• RuntimeError: если выполнение конвейера данных завершается сбоем.

Пример:

spec = GraphSpecBuilder.start(context=ctx).add_node(...).done()
graph = Graph.prepare(spec)
# Inspect results before publishing
graph.nodes.show()
graph.publish()
graph.query("MATCH (n) RETURN n")

build

Graph.build(spec: GraphSpec) -> Graph

Создайте граф на основе путем GraphSpec подготовки данных и публикации. Внутренние вызовы Graph.prepare(spec) и попытки graph.publish(). В отличие от вызова этих двух методов по отдельности, ошибки публикации перехватываются — возвращаемый граф имеет build_status.status == "prepared" и build_status.api_error задает вместо того, чтобы вызывать.

Параметры:

• spec (GraphSpec): спецификация графа для сборки

Возвращает:

• Graph: экземпляр Graph с spec присоединенным и build_status заполненным

Поднимает:

• ValueError: если спецификация не имеет конвейера данных или контекста выполнения.
• RuntimeError: если выполнение конвейера данных завершается сбоем.

Пример:

spec = GraphSpecBuilder.start(context=ctx).add_node(...).done()
graph = Graph.build(spec)
print(graph.build_status.status)  # "published" or "prepared" (None if neither ran)
graph.query("MATCH (n) RETURN n")

Свойства

nodes

def nodes() -> Optional[DataFrame]

Получение кадра данных узлов. Делегирует при self.spec.nodes присоединении спецификации; возвращает в None противном случае.

edges

def edges() -> Optional[DataFrame]

Получение ребер DataFrame. Делегирует при self.spec.edges присоединении спецификации; возвращает в None противном случае.

schema

def schema() -> Optional[GraphSchema]

Получение схемы графа. Делегирует при self.spec.get_schema() присоединении спецификации; возвращает в None противном случае.

Методы

query

def query(query_string: str, query_language: str = "GQL") -> QueryResult

Выполните запрос к экземпляру графа с помощью GQL.

Параметры:

• query_string (str): строка запроса Graph (язык GQL)
• query_language (str, default="GQL"): язык запросов

Возвращает:

• QueryResult: объект, содержащий узлы, ребра и метаданные.

Поднимает:

• ValueError: если отсутствует сеанс ExecutionContext или Spark.
• RuntimeError: если инициализация клиента или выполнение запроса завершается сбоем.

Пример:

result = graph.query("MATCH (u:user) WHERE u.age > 30 RETURN u")
result.show()

reachability

def reachability(
    *,
    source_property_value: str = None,
    target_property_value: str = None,
    source_property: Optional[str] = None,
    participating_source_node_labels: Optional[List[str]] = None,
    target_property: Optional[str] = None,
    participating_target_node_labels: Optional[List[str]] = None,
    participating_edge_labels: Optional[List[str]] = None,
    is_directional: bool = True,
    min_hop_count: int = 1,
    max_hop_count: int = 4,
    shortest_path: bool = False,
    max_results: int = 500
) -> QueryResult

[! ПРИМЕЧАНИЕ] reachability(query_input=ReachabilityQueryInput(...)) по-прежнему принимается, но выдает DeprecationWarning и будет удален в будущей версии.

Анализ доступности между исходным и целевым узлами.

Параметры:

• source_property_value (str): значение, соответствующее свойству источника (проверено во время выполнения). Должен быть указан, если не используется query_input)
• target_property_value (str): значение, соответствующее целевому свойству (проверяется во время выполнения). Должен быть указан, если не используется query_input)
• source_property (Необязательно[str]): имя свойства для фильтрации исходных узлов
• participating_source_node_labels (Необязательно[List[str]]): метки узла для рассмотрения в качестве источников
• target_property (Необязательно[str]): имя свойства для фильтрации целевых узлов
• participating_target_node_labels (Необязательно[List[str]]): метки узла, которые следует рассматривать в качестве целевых объектов
• participating_edge_labels (Необязательно[List[str]]): пограничные метки для обхода
• is_directional (bool): являются ли ребра направленными (по умолчанию: True)
• min_hop_count (int): минимальное количество прыжков (по умолчанию: 1)
• max_hop_count (int): максимальное количество прыжков (по умолчанию: 4)
• shortest_path (bool): возвращает только кратчайшие пути (по умолчанию: False)
• max_results (int): максимальное количество результатов (по умолчанию: 500)

Поднимает:

• ValueError: если source_property_value или target_property_value отсутствует, min_hop_count < 1, max_hop_count < min_hop_countили max_results < 1
• RuntimeError: если инициализация клиента или выполнение запроса завершается сбоем.

Возвращает:

• QueryResult: содержит пути доступности

Пример:

result = graph.reachability(
    source_property_value="user-001",
    target_property_value="device-003")
result.show()

k_hop

def k_hop(
    *,
    source_property: Optional[str] = None,
    source_property_value: Optional[str] = None,
    participating_source_node_labels: Optional[List[str]] = None,
    target_property: Optional[str] = None,
    target_property_value: Optional[str] = None,
    participating_target_node_labels: Optional[List[str]] = None,
    participating_edge_labels: Optional[List[str]] = None,
    is_directional: bool = True,
    min_hop_count: int = 1,
    max_hop_count: int = 4,
    shortest_path: bool = False,
    max_results: int = 500
) -> QueryResult

Выполните анализ k-hop из заданного исходного узла.

Параметры:

• То же, что и reachability

Проверки:

• Должен быть указан по крайней мере один из source_property_value или target_property_value

Поднимает:

• ValueError: если ни source_property_value , ни target_property_value не указано, либо если нарушаются числовые ограничения (то же самое, что reachability)
• RuntimeError: если инициализация клиента или выполнение запроса завершается сбоем.

Возвращает:

• QueryResult: содержит результаты k-hop

Пример:

result = graph.k_hop(source_property_value="user-001")
result.show()

blast_radius

def blast_radius(
    *,
    source_property_value: str = None,
    target_property_value: str = None,
    source_property: Optional[str] = None,
    participating_source_node_labels: Optional[List[str]] = None,
    target_property: Optional[str] = None,
    participating_target_node_labels: Optional[List[str]] = None,
    participating_edge_labels: Optional[List[str]] = None,
    is_directional: bool = True,
    min_hop_count: int = 1,
    max_hop_count: int = 4,
    shortest_path: bool = False,
    max_results: int = 500
) -> QueryResult

Выполните анализ радиуса взрыва от исходного узла до целевого узла.

Параметры:

• source_property_value (str): значение, определяющее исходный узел (проверено во время выполнения). Должен быть указан, если не используется query_input)
• target_property_value (str): значение, определяющее целевой узел (проверено во время выполнения). Должен быть указан, если не используется query_input)
• Другие параметры: то же самое, что и reachability

Поднимает:

• ValueError: если source_property_value или target_property_value отсутствует, или если нарушены числовые ограничения (то же, что и reachability)
• RuntimeError: если инициализация клиента или выполнение запроса завершается сбоем.

Возвращает:

• QueryResult: содержит результаты радиуса взрыва.

Пример:

result = graph.blast_radius(
    source_property_value="user-003",
    target_property_value="device-003",
    min_hop_count=1)
result.show()

centrality

def centrality(
    *,
    participating_source_node_labels: Optional[List[str]] = None,
    participating_target_node_labels: Optional[List[str]] = None,
    participating_edge_labels: Optional[List[str]] = None,
    threshold: int = 3,
    centrality_type: CentralityType = None,
    max_paths: int = 1000000,
    is_directional: bool = True,
    min_hop_count: int = 1,
    max_hop_count: int = 4,
    shortest_path: bool = False,
    max_results: int = 500
) -> QueryResult

Выполните анализ централизации графа.

Параметры:

• participating_source_node_labels (Необязательно[List[str]]): метки исходного узла
• participating_target_node_labels (Необязательно[List[str]]): метки целевого узла
• participating_edge_labels (Необязательно[List[str]]): пограничные метки для обхода
• threshold (int): минимальная оценка централизации (по умолчанию: 3); должна быть неотрицательной.
• centrality_type (CentralityType): CentralityType.Node или CentralityType.Edge (по умолчанию: None, возвращается к CentralityType.Node)
• max_paths (int): максимальное число путей для рассмотрения (по умолчанию: 1000000; 0 = все пути); должно быть неотрицательно.
• is_directional (bool): являются ли ребра направленными (по умолчанию: True)
• min_hop_count (int): минимальное число прыжков (по умолчанию: 1); должно быть ≥ 1
• max_hop_count (int): максимальное число прыжков (по умолчанию : 4); должно быть ≥ min_hop_count
• shortest_path (bool): возвращает только кратчайшие пути (по умолчанию: False)
• max_results (int): максимальное количество результатов (по умолчанию : 500); должно быть ≥ 1

Поднимает:

• ValueError: если threshold < 0, max_paths < 0, min_hop_count < 1, max_hop_count < min_hop_countили max_results < 1
• RuntimeError: если инициализация клиента или выполнение запроса завершается сбоем.

Возвращает:

• QueryResult: содержит метрики централизации.

Пример:

result = graph.centrality(
    participating_source_node_labels=["user", "device"],
    participating_target_node_labels=["device", "user"],
    participating_edge_labels=["sign_in"],
    is_directional=False)
result.show()

ranked

def ranked(
    *,
    rank_property_name: str = None,
    threshold: int = 0,
    max_paths: int = 1000000,
    decay_factor: float = 1,
    is_directional: bool = True,
    min_hop_count: int = 1,
    max_hop_count: int = 4,
    shortest_path: bool = False,
    max_results: int = 500
) -> QueryResult

Выполните ранжированный анализ на графе.

Параметры:

• rank_property_name (str): имя свойства, используемое для ранжирования (проверяется во время выполнения). Должен быть указан, если не используется query_input)
• threshold (int): только пути возврата выше этого веса (по умолчанию: 0); должны быть неотрицательно
• max_paths (int): максимальное число путей для рассмотрения (по умолчанию: 1000000; 0 = все пути); должно быть неотрицательно.
• decay_factor (float): упадок ранга на шаг; 2 означает, что половины (по умолчанию: 1); должно быть не отрицательным
• is_directional (bool): являются ли ребра направленными (по умолчанию: True)
• min_hop_count (int): минимальное число прыжков (по умолчанию: 1); должно быть ≥ 1
• max_hop_count (int): максимальное число прыжков (по умолчанию : 4); должно быть ≥ min_hop_count
• shortest_path (bool): возвращает только кратчайшие пути (по умолчанию: False)
• max_results (int): максимальное количество результатов (по умолчанию : 500); должно быть ≥ 1

Поднимает:

• ValueError: если rank_property_name параметр отсутствует, threshold < 0, max_paths < 0, decay_factor < 0, min_hop_count < 1, max_hop_count < min_hop_count, или max_results < 1
• RuntimeError: если инициализация клиента или выполнение запроса завершается сбоем.

Возвращает:

• QueryResult: содержит ранжированные узлы и ребра.

Пример:

result = graph.ranked(
    rank_property_name="risk_score",
    threshold=5,
    decay_factor=2)
result.show()

to_graphframe

def to_graphframe(column_mapping: Optional[Dict[str, str]] = None) -> GraphFrame

Преобразование всего графа в GraphFrame. Использует данные спецификаций, если они доступны; в противном случае считывает данные из таблиц озера.

Параметры:

• column_mapping (Dict[str, str], необязательно): настраиваемое сопоставление столбцов

Возвращает:

• GraphFrame: объект GraphFrame со всеми вершинами и ребрами

Пример:

gf = graph.to_graphframe()

show

def show() -> None

Отображение сведений о графе. Делегирует в spec.show() для полнофункционированного отображения при присоединении спецификации; в противном случае выводит минимальные сведения.

show_schema

def show_schema() -> None

Отображение схемы графа. Делегирует значение при spec.show_schema() присоединении спецификации; выводит сообщение, указывающее, что схема недоступна в противном случае.

publish (новое в версии 0.3.3)

def publish() -> Graph

Зарегистрируйте граф с помощью API, сделав его запрашиваемым. Вызовите его после Graph.prepare() (или на любом Graph экземпляре с присоединенной спецификацией), чтобы опубликовать экземпляр графа.

Возвращает:

• Graph: self for method chaining

Поднимает:

• ValueError: если спецификация не присоединена или контекст отсутствует.
• RuntimeError: если публикация завершается ошибкой

Пример:

graph = Graph.prepare(spec)
graph.publish()
# Now the graph is queryable
graph.query("MATCH (n) RETURN n")

BuildStatus

Класс данных, несущий метаданные Graph.build() из операции.

Fields

Пути построения

GraphSpecBuilder.start(...).done()  →  GraphSpec             (spec only, no graph yet)
Graph.get(name, context)            →  Graph (spec=None, build_status=None)
Graph.prepare(spec)                 →  Graph (spec=spec, build_status.status="prepared")
graph.publish()                     →  Graph (build_status.status="published")
Graph.build(spec)                   →  Graph (prepare + publish in one step)

Пример:

graph = Graph.build(spec)
if graph.build_status.status == "published":
    print("Graph prepared and published successfully")
elif graph.build_status.status == "prepared":
    print(f"Prepare succeeded but publish failed: {graph.build_status.api_error}")
elif graph.build_status.status is None:
    print("Neither prepare nor publish has run")

Построитель узлов

NodeBuilderInitial

Начальное состояние для построителя узлов: доступны только методы источника данных.

constructor

NodeBuilderInitial(alias: str, graph_builder: GraphSpecBuilder)

Примечание: Обычно создается с помощью GraphSpecBuilder.add_node(), а не создается напрямую.

Методы

from_table

def from_table(self, table_name: str, database: Optional[str] = None, time_generated_start: Optional[str] = None, time_generated_end: Optional[str] = None) -> NodeBuilderSourceSet

Задайте таблицу в качестве источника данных с интеллектуальным разрешением базы данных.

Параметры:

• table_name (str): имя таблицы (обязательно)
• database (str, необязательно): явное имя базы данных (имеет приоритет над контекстом по умолчанию)
• time_generated_start (str, необязательный) и ( time_generated_end str, необязательный): необязательные поля метки времени при использовании вместе активирует оптимизированное разностное чтение. Мы рекомендуем использовать каноническую форму: "гггг-ММ-дд HH:mm:ss".

Возвращает:

• NodeBuilderSourceSet: построитель для дальнейшей настройки

Поднимает:

• ValueError: если таблица не найдена или найдено несколько конфликтующих таблиц.

Порядок разрешения базы данных:

1. Явный database параметр (наивысший приоритет)
2. ExecutionContext.default_database
3. Поиск по всем базам данных (с обнаружением конфликтов)

Пример:

end = datetime.now(timezone.utc)
start = end - timedelta(hours=24)

builder.add_node("user").from_table("SigninLogs", database="security_db", time_generated_start = start, time_generated_end = end)

from_dataframe

def from_dataframe(dataframe: DataFrame) -> NodeBuilderSourceSet

Задайте Spark DataFrame в качестве источника данных.

Параметры:

• dataframe (DataFrame): Кадр данных Spark

Возвращает:

• NodeBuilderSourceSet: построитель для дальнейшей настройки

Пример:

# For Standard Read
df = spark.read.table("users", WORKSPACE_NAME)

# Optional Optimized Read - You could use the below pattern to activate Optimized Delta Reading
end = datetime.now(timezone.utc)
start = end - timedelta(hours=24)

df = spark.read.table("users", WORKSPACE_NAME, time_generated_start = start, time_generated_end = end )

builder.add_node("user").from_dataframe(df)

NodeBuilderSourceSet

Построитель узлов после установки источника данных: доступные методы конфигурации.

constructor

NodeBuilderSourceSet(alias: str, graph_builder: GraphSpecBuilder, source_step: DataInputETLStep)

Примечание: Создано внутренне методом источника NodeBuilderInitial.

Методы

with_time_range

def with_time_range(
    time_column: str,
    start_time: Optional[Union[str, datetime]] = None,
    end_time: Optional[Union[str, datetime]] = None,
    lookback_hours: Optional[float] = None
) -> NodeBuilderSourceSet

Примените фильтрацию диапазона времени к источнику данных узла.

Параметры:

• time_column (str): имя столбца, содержащего данные метки времени (обязательно)
• start_time (str или datetime, необязательно): начальная дата ('10/20/25', '2025-10-20' или объект datetime)
• end_time (str или datetime, необязательно): дата окончания (те же форматы, что и start_time)
• lookback_hours (float, необязательный): часы, чтобы оглянуться назад

Возвращает:

• NodeBuilderSourceSet: self for method chaining

Поднимает:

• ValueError: если столбец time не найден в исходной схеме

Логика диапазона времени:

1. Если start_time и end_time предоставлены, используйте их напрямую.
2. Если только lookback_hours указано: end=now, start=now-lookback_hours
3. Если ничего не указано: нет фильтрации по времени
4. Если начальная и конечная lookback_hours: начало и конец имеют приоритет

Пример:

# Explicit date range
builder.add_node("user").from_table("SigninLogs") \
    .with_time_range(time_column="TimeGenerated", start_time="2025-01-01", end_time="2025-01-31")

# Lookback window
builder.add_node("user").from_table("SigninLogs") \
    .with_time_range(time_column="TimeGenerated", lookback_hours=24)

with_label

def with_label(label: str) -> NodeBuilderSourceSet

Установите метку узла (по умолчанию используется псевдоним, если он не вызывается).

Параметры:

• label (str): метка узла

Возвращает:

• NodeBuilderSourceSet: self for method chaining

Поднимает:

• ValueError: если метка уже задана

Пример:

builder.add_node("u").from_table("Users").with_label("user")

with_columns

def with_columns(
    *columns: str,
    key: str,
    display: str
) -> NodeBuilderSourceSet

Настройте столбцы с обязательным обозначением ключа и отображения.

Параметры:

• *columns (str): имена столбцов для включения (по крайней мере один обязательный)
• key (str): имя столбца для пометки как ключа (обязательно, должно быть в столбцах)
• display (str): имя столбца, которое помечается как отображаемое значение (обязательно, должно быть в столбцах, может совпадать с ключом)

Возвращает:

• NodeBuilderSourceSet: self for method chaining

Поднимает:

• ValueError: если проверка завершается ошибкой (дублирование столбцов, отсутствие ключа или отображения и т. д.)

Примечания.

• Свойства автоматически создаются на основе типов столбцов

• Столбец фильтра времени добавляется автоматически, если он указан

• Типы свойств автоматически выводятся из исходной схемы

• См. раздел Ограничения

Пример:

builder.add_node("user").from_table("Users") \
    .with_columns("id", "name", "email", "created_at", key="id", display="name")

add_node

def add_node(alias: str) -> NodeBuilderInitial

Завершите этот узел и начните создание другого узла.

Параметры:

• alias (str): псевдоним для нового узла

Возвращает:

• NodeBuilderInitial: новый построитель узлов

Пример:

builder.add_node("user").from_table("Users") \
    .with_columns("id", "name", key="id", display="name") \
    .add_node("device")

add_edge

def add_edge(alias: str) -> EdgeBuilderInitial

Завершите этот узел и начните создание ребра.

Параметры:

• alias (str): псевдоним для края

Возвращает:

• EdgeBuilderInitial: новый построитель edge

Пример:

builder.add_node("user").from_table("Users") \
    .with_columns("id", "name", key="id", display="name") \
    .add_edge("accessed")

done

def done() -> GraphSpec

Завершите работу этого узла и завершите спецификацию графа.

Возвращает:

• GraphSpec: полная спецификация графа

Пример:

graph_spec = builder.add_node("user").from_table("Users") \
    .with_columns("id", "name", key="id", display="name") \
    .done()

Конструкторы Edge

EdgeBuilderInitial

Начальное состояние для пограничного построителя: доступны только методы источника данных.

constructor

EdgeBuilderInitial(alias: str, graph_builder: GraphSpecBuilder)

Примечание: Обычно создается с помощью GraphSpecBuilder.add_edge(), а не создается напрямую.

Методы

from_table

def from_table(table_name: str, database: Optional[str] = None) -> EdgeBuilderSourceSet

Задайте таблицу в качестве источника данных с интеллектуальным разрешением базы данных.

Параметры:

• table_name (str): имя таблицы (обязательно)
• database (str, необязательно): явное имя базы данных

Возвращает:

• EdgeBuilderSourceSet: построитель для дальнейшей настройки

Поднимает:

• ValueError: если таблица не найдена или найдено несколько конфликтующих таблиц.

Пример:

builder.add_edge("accessed").from_table("AccessLogs")

from_dataframe

def from_dataframe(dataframe: DataFrame) -> EdgeBuilderSourceSet

Задайте Spark DataFrame в качестве источника данных.

Параметры:

• dataframe (DataFrame): Кадр данных Spark

Возвращает:

• EdgeBuilderSourceSet: построитель для дальнейшей настройки

Пример:

df = spark.read.table("access_logs")
builder.add_edge("accessed").from_dataframe(df)

EdgeBuilderSourceSet

Построитель Edge после установки источника данных: доступные методы конфигурации.

constructor

EdgeBuilderSourceSet(alias: str, graph_builder: GraphSpecBuilder, source_step: DataInputETLStep)

Примечание: Внутреннее создание с помощью исходных методов EdgeBuilderInitial.

Методы

with_label

def with_label(label: str) -> EdgeBuilderSourceSet

Задайте тип или метку отношения ребер (по умолчанию используется псевдоним, если он не вызывается).

Параметры:

• label (str): метка edge

Возвращает:

• EdgeBuilderSourceSet: self for method chaining

Поднимает:

• ValueError: если метка уже задана

Пример:

builder.add_edge("rel").from_table("AccessLogs").with_label("ACCESSED")

edge_label

def edge_label(label: str) -> EdgeBuilderSourceSet

Задайте тип или метку отношения ребер (по умолчанию используется псевдоним, если он не вызывается).

Параметры:

• label (str): метка edge

Возвращает:

• EdgeBuilderSourceSet: self for method chaining

Поднимает:

• ValueError: если метка уже задана

Пример:

builder.add_edge("acc").from_table("AccessLogs").edge_label("accessed")

source

def source(id_column: str, node_type: str) -> EdgeBuilderSourceSet

Задайте исходный узел со столбцом идентификатора и меткой.

Параметры:

• id_column (str): имя столбца, содержащего идентификатор исходного узла.
• node_type (str): метка исходного узла

Возвращает:

• EdgeBuilderSourceSet: self for method chaining

Поднимает:

• ValueError: если источник уже задан

Пример:

builder.add_edge("accessed").from_table("AccessLogs") \
    .source(id_column="user_id", node_type="user")

target

def target(id_column: str, node_type: str) -> EdgeBuilderSourceSet

Задайте целевой узел со столбцом идентификатора и меткой.

Параметры:

• id_column (str): имя столбца, содержащего идентификатор целевого узла.
• node_type (str): метка целевого узла

Возвращает:

• EdgeBuilderSourceSet: self for method chaining

Поднимает:

• ValueError: если целевой объект уже задан

Пример:

builder.add_edge("accessed").from_table("AccessLogs") \
    .source(id_column="user_id", node_type="user") \
    .target(id_column="device_id", node_type="device")

with_time_range

def with_time_range(
    time_column: str,
    start_time: Optional[Union[str, datetime]] = None,
    end_time: Optional[Union[str, datetime]] = None,
    lookback_hours: Optional[float] = None
) -> EdgeBuilderSourceSet

Примените фильтрацию по диапазону времени к источнику данных края.

Параметры:

• time_column (str): имя столбца, содержащего данные метки времени (обязательно)
• start_time (str или datetime, необязательно): дата начала
• end_time (str или datetime, необязательно): дата окончания
• lookback_hours (float, необязательный): часы, чтобы оглянуться назад

Возвращает:

• EdgeBuilderSourceSet: self for method chaining

Поднимает:

• ValueError: если столбец time не найден в исходной схеме

Пример:

builder.add_edge("accessed").from_table("AccessLogs") \
    .with_time_range(time_column="TimeGenerated", lookback_hours=48)

with_columns

def with_columns(
    *columns: str,
    key: str,
    display: str
) -> EdgeBuilderSourceSet

Настройте столбцы с обязательным обозначением ключа и отображения.

Параметры:

• *columns (str): имена столбцов для включения (по крайней мере один обязательный)
• key (str): имя столбца для пометки как ключа (обязательно, должно быть в столбцах)
• display (str): имя столбца, которое помечается как отображаемое значение (обязательно, должно быть в столбцах)

Возвращает:

• EdgeBuilderSourceSet: self for method chaining

Поднимает:

• ValueError: если проверка завершается сбоем.

Пример:

builder.add_edge("accessed").from_table("AccessLogs") \
    .source(id_column="user_id", node_type="user") \
    .target(id_column="device_id", node_type="device") \
    .with_columns("id", "location", "status", key="id", display="location")

add_node

def add_node(alias: str) -> NodeBuilderInitial

Завершите эту границу и начните создание узла.

Параметры:

• alias (str): псевдоним для нового узла

Возвращает:

• NodeBuilderInitial: новый построитель узлов

add_edge

def add_edge(alias: str) -> EdgeBuilderInitial

Завершите этот ребро и начните создание другого ребра.

Параметры:

• alias (str): псевдоним для нового края

Возвращает:

• EdgeBuilderInitial: новый построитель edge

Пример:

builder.add_edge("accessed").from_table("AccessLogs") \
    .source(id_column="user_id", node_type="user") \
    .target(id_column="device_id", node_type="device") \
    .with_columns("id", "location", key="id", display="location") \
    .add_edge("connected_to")

done

def done() -> GraphSpec

Завершите эту границу и завершите спецификацию графа.

Возвращает:

• GraphSpec: полная спецификация графа

Классы схемы

GraphDefinitionReference

Ссылка на определение графа с именем и версией.

constructor

GraphDefinitionReference(
    fully_qualified_name: str,
    version: str
)

Параметры:

• fully_qualified_name (str): полное имя графа, на который ссылается ссылка.
• version (str): версия графа, на который ссылается ссылка

Поднимает:

• ValueError: если fully_qualified_name или версия пуста.

Методы

to_dict

def to_dict() -> Dict[str, Any]

Сериализация в словарь.

Возвращает:

• Dict[str, Any]: сериализованная ссылка

Property

Определение свойства с типобезопасного интерфейса.

constructor

Property(
    name: str,
    property_type: PropertyType,
    is_non_null: bool = False,
    description: str = "",
    is_key: bool = False,
    is_display_value: bool = False,
    is_internal: bool = False
)

Параметры:

• name (str): имя свойства
• property_type (PropertyType): тип данных свойства
• is_non_null (bool, default=False): требуется ли свойство
• description (str, default=""): описание свойства
• is_key (bool, default=False): является ли свойство ключом
• is_display_value (bool, default=False): является ли свойство отображаемым значением
• is_internal (bool, default=False): является ли свойство внутренним

Поднимает:

• ValueError: если имя пусто или проверка завершается ошибкой.

Методы класса

key

@classmethod
Property.key(
    name: str,
    property_type: PropertyType,
    description: str = "",
    is_non_null: bool = False
) -> Property

Создайте свойство ключа с общими параметрами (is_key=True, is_display_value=True).

display

@classmethod
Property.display(
    name: str,
    property_type: PropertyType,
    description: str = "",
    is_non_null: bool = False
) -> Property

Создайте свойство отображаемого значения (is_display_value=True).

Методы

describe

def describe(text: str) -> Property

Добавьте описание свободно.

Параметры:

• text (str): текст описания

Возвращает:

• Property: self for method chaining

to_dict

def to_dict() -> Dict[str, Any]

Сериализация свойства в словарь с помощью ключей заметок с префиксом @.

Возвращает:

• Dict[str, Any]: сериализованное свойство

to_gql

def to_gql() -> str

Создание определения свойства GQL.

Возвращает:

• str: представление строки GQL

EdgeNode

Ссылка на узел, используемая в определениях ребер.

constructor

EdgeNode(
    alias: Optional[str] = None,
    labels: List[str] = []
)

Параметры:

• alias (str, необязательно): псевдоним узла (автоматически задается первая метка, если нет или пуст)
• labels (List[str]): метки узла (по крайней мере одна обязательная)

Поднимает:

• ValueError: если список меток пуст.
• TypeError: если метки не являются строками

Автоматическая мутация:

• Если псевдоним равен None или пуст, для него устанавливается первая метка.

Методы

to_dict

def to_dict() -> Dict[str, Any]

Сериализация в словарь.

Возвращает:

• Dict[str, Any]: ссылка на сериализованный пограничный узел

сайте

Определение узла с типобезопасной интерфейсом.

constructor

Node(
    alias: str = "",
    labels: List[str] = [],
    implies_labels: List[str] = [],
    properties: List[Property] = [],
    description: str = "",
    entity_group: str = "",
    dynamic_labels: bool = False,
    abstract_edge_aliases: bool = False
)

Параметры:

• alias (str, default=""): псевдоним узла (автоматически задается первая метка, если пуст)
• labels (List[str]): метки узла (по крайней мере одна обязательная)
• implies_labels (List[str], default=[]): подразумеваемые метки
• properties (List[Property], default=[]): свойства узла
• description (str, default=""): описание узла
• entity_group (str, default=""): имя группы сущностей
• dynamic_labels (bool, default=False): имеет ли узел динамические метки
• abstract_edge_aliases (bool, default=False): использует ли узел абстрактные псевдонимы ребер.

Поднимает:

• ValueError: если проверка завершается ошибкой (без меток, без свойства ключа, без свойства отображения и т. д.)

Автоматическая мутация:

• Если псевдоним пуст, для него устанавливается первая метка.
• Если entity_group пуст, задается основная метка.

Методы

get_primary_label

def get_primary_label() -> Optional[str]

Получите основную (первую) метку.

Возвращает:

• str или None: основная метка или Нет, если меток нет.

get_entity_group_name

def get_entity_group_name() -> str

Получите имя группы сущностей или откат к первичной метке.

Возвращает:

• str: имя группы сущностей

get_primary_key_property_name

def get_primary_key_property_name() -> Optional[str]

Получите имя свойства первичного ключа.

Возвращает:

• str или None: имя свойства первичного ключа

get_properties

def get_properties() -> Dict[str, Property]

Получение свойств в виде словаря для удобного доступа.

Возвращает:

• Dict[str, Property]: свойства с ключом по имени

get_property

def get_property(name: str) -> Optional[Property]

Получение определенного свойства по имени.

Параметры:

• name (str): имя свойства

Возвращает:

• Property или None: свойство , если оно найдено

add_property

def add_property(prop: Property) -> None

Добавьте свойство в этот узел.

Параметры:

• prop (Свойство): добавляемое свойство

Поднимает:

• ValueError: если имя свойства дублируется.

is_dynamically_labeled

def is_dynamically_labeled() -> bool

Проверьте, есть ли у узла динамические метки.

Возвращает:

• bool: значение True, если динамические метки включены.

is_abstract_edge_node_aliases

def is_abstract_edge_node_aliases() -> bool

Проверьте, использует ли узел абстрактные пограничные псевдонимы узла.

Возвращает:

• bool: значение true, если включены абстрактные псевдонимы ребер.

describe

def describe(text: str) -> Node

Добавьте описание свободно.

Параметры:

• text (str): текст описания

Возвращает:

• Node: self for method chaining

to_dict

def to_dict() -> Dict[str, Any]

Сериализация узла в словарь.

Возвращает:

• Dict[str, Any]: сериализованный узел

to_gql

def to_gql() -> str

Создайте определение узла GQL.

Возвращает:

• str: представление строки GQL

Поднимает:

• ValueError: если в узле отсутствуют обязательные поля для GQL.

Методы класса

create

@classmethod
Node.create(
    alias: str,
    labels: List[str],
    properties: List[Property],
    description: str = "",
    entity_group: str = "",
    **kwargs
) -> Node

Создайте узел со всеми обязательными полями.

Параметры:

• alias (str): псевдоним узла
• labels (List[str]): метки узла
• properties (List[Property]): Свойства узла
• description (str, default=""): описание узла
• entity_group (str, default=""): имя группы сущностей

Возвращает:

• Node: новый экземпляр узла

Microsoft Edge

Определение edge с типобезопасными интерфейсами.

constructor

Edge(
    relationship_type: str,
    source_node_label: str,
    target_node_label: str,
    direction: EdgeDirection = EdgeDirection.DIRECTED_RIGHT,
    properties: List[Property] = [],
    description: str = "",
    entity_group: str = "",
    dynamic_type: bool = False
)

Параметры:

• relationship_type (str): тип отношения Edge (например, "FOLLOWS", "OWNS")
• source_node_label (str): метка исходного узла
• target_node_label (str): метка целевого узла
• direction (EdgeDirection, default=DIRECTED_RIGHT): направление edge
• properties (List[Property], default=[]): свойства edge
• description (str, default=""): описание edge
• entity_group (str, default=""): имя группы сущностей
• dynamic_type (bool, default=False): имеет ли ребро динамический тип

Поднимает:

• ValueError: если проверка завершается сбоем.

Автоматическая мутация:

• labels список автоматически заполняется [relationship_type]
• Если entity_group пуст, для него задано значение relationship_type

Свойства

edge_type

def edge_type() -> str

Псевдоним обратной совместимости для relationship_type.

Возвращает:

• str: тип связи

Методы

get_entity_group_name

def get_entity_group_name() -> str

Получите имя группы сущностей или откат к типу связи.

Возвращает:

• str: имя группы сущностей

is_dynamic_type

def is_dynamic_type() -> bool

Проверьте, имеет ли ребро динамический тип.

Возвращает:

• bool: значение True, если динамический тип

add_property

def add_property(edge_property: Property) -> None

Добавьте свойство к этому краю.

Параметры:

• edge_property (Свойство): добавляемое свойство

describe

def describe(text: str) -> Edge

Добавьте описание свободно.

Параметры:

• text (str): текст описания

Возвращает:

• Edge: self for method chaining

to_dict

def to_dict() -> Dict[str, Any]

Сериализация edge в словарь.

Возвращает:

• Dict[str, Any]: сериализованный край

to_gql

def to_gql() -> str

Создайте определение границЫ GQL.

Возвращает:

• str: представление строки GQL

Методы класса

create

Edge.create(
    relationship_type: str,
    source_node_label: str,
    target_node_label: str,
    properties: List[Property] = None,
    description: str = "",
    entity_group: str = "",
    **kwargs
) -> Edge

Создайте ребро со всеми обязательными полями.

Параметры:

• relationship_type (str): тип связи Edge
• source_node_label (str): метка исходного узла
• target_node_label (str): метка целевого узла
• properties (List[Property], необязательный): свойства edge
• description (str, default=""): описание edge
• entity_group (str, default=""): имя группы сущностей

Возвращает:

• Edge: новый пограничный экземпляр

GraphSchema

Определение схемы графа с типобезопасными интерфейсами.

constructor

GraphSchema(
    name: str,
    nodes: List[Node] = [],
    edges: List[Edge] = [],
    base_graphs: List[GraphSchema] = [],
    description: str = "",
    version: str = "1.0",
    fully_qualified_name: str = "",
    namespace: str = ""
)

Параметры:

• name (str): имя схемы графа
• nodes (List[Node], default=[]): определения узлов
• edges (List[Edge], default=[]): определения edge
• base_graphs (List[GraphSchema], default=[]): схемы базовых графов
• description (str, default=""): описание схемы
• version (str, default="1.0"): Версия схемы
• fully_qualified_name (str, default=""): полное имя
• namespace (str, default=""): пространство имен

Поднимает:

• ValueError: если проверка завершается ошибкой (повторяющиеся псевдонимы, ребра ссылают на несуществующие узлы и т. д.)

Методы

get_fully_qualified_name

def get_fully_qualified_name() -> str

Получите полное имя.

Возвращает:

• str: полное имя

get_namespace

def get_namespace() -> str

Возвращает пространство имен из полного имени или возвращает значение по умолчанию.

Возвращает:

• str:Пространства имен

get_version

def get_version() -> str

Получить версию.

Возвращает:

• str: строка версии

get_node

def get_node(label_or_alias: str) -> Optional[Node]

Получение узла по меткам или псевдонимам.

Параметры:

• label_or_alias (str): метка узла или псевдоним

Возвращает:

• Node или None: узел, если он найден

get_edge

def get_edge(name: str) -> Optional[Edge]

Получение края по имени или типу.

Параметры:

• name (str): тип связи Edge

Возвращает:

• Edge или None: edge , если найдено

add_node

def add_node(node: Node) -> None

Добавьте узел в этот граф.

Параметры:

• node (Узел): узел для добавления

Поднимает:

• ValueError: если псевдоним узла дублируется.

add_edge

def add_edge(edge: Edge) -> None

Добавьте ребро в этот граф.

Параметры:

• edge (Edge): edge для добавления

Поднимает:

• ValueError: если тип ребра дублируется.

include_graph

def include_graph(fully_qualified_name: str, version: str) -> GraphSchema

Добавление графа (fluent API).

Параметры:

• fully_qualified_name (str): полное имя графа для включения
• version (str): версия графа для включения

Возвращает:

• GraphSchema: self for method chaining

get_included_graph_references

def get_included_graph_references() -> List[GraphDefinitionReference]

Получение списка включенных ссылок на графы.

Возвращает:

• List[GraphDefinitionReference]: список ссылок на определения графа

describe

def describe(text: str) -> GraphSchema

Добавьте описание свободно.

Параметры:

• text (str): текст описания

Возвращает:

• GraphSchema: self for method chaining

to_dict

def to_dict() -> Dict[str, Any]

Сериализация схемы в словарь.

Возвращает:

• Dict[str, Any]: сериализованная схема

to_json

def to_json(indent: int = 2) -> str

Создание представления JSON.

Параметры:

• indent (int, default=2): уровень отступа JSON

Возвращает:

• str: строка JSON

to_gql

def to_gql() -> str

Создание определения схемы GQL.

Возвращает:

• str: представление строки GQL

Методы класса

create

@classmethod
GraphSchema.create(
    name: str,
    nodes: List[Node] = None,
    edges: List[Edge] = None,
    description: str = "",
    version: str = "1.0",
    **kwargs
) -> GraphSchema

Создайте схему графа со всеми обязательными полями.

Параметры:

• name (str): имя схемы графа
• nodes (List[Node], необязательно): Определения узлов
• edges (List[Edge], необязательно): определения edge
• description (str, default=""): описание схемы
• version (str, default="1.0"): Версия схемы

Возвращает:

• GraphSchema: новый экземпляр схемы графа

Входные классы запросов

Классы данных, представляющие входные параметры для предопределенных запросов графов.

QueryInputBase

Базовый класс для всех входных параметров запроса.

Методы

to_json_payload

def to_json_payload() -> Dict[str, Any]

Преобразуйте входные параметры в словарь для отправки API.

Возвращает:

• Dict[str, Any]: представление входных параметров в словаре

validate

def validate() -> None

Проверьте входные параметры.

Поднимает:

• ValueError: если входные параметры недопустимы.

ReachabilityQueryInput

Входные параметры для запроса на доступность между исходным и целевым узлами. Наследует , от ReachabilityQueryInputBase которого наследуется от QueryInputBase.

Fields

Проверки:

• source_property_value является обязательным
• target_property_value является обязательным

Пример:

# Preferred: keyword arguments (no import needed)
result = graph.reachability(
    source_property="UserId",
    source_property_value="user123",
    target_property="DeviceId",
    target_property_value="device456",
    participating_edge_labels=["accessed", "connected_to"],
    shortest_path=True
)

# DEPRECATED — will be removed in a future version. Use keyword arguments above.
from sentinel_graph.builders.query_input import ReachabilityQueryInput
result = graph.reachability(query_input=ReachabilityQueryInput(
    source_property_value="user123",
    target_property_value="device456"
))

K_HopQueryInput

Входные параметры для запроса k-hop из заданного исходного узла. Наследует от ReachabilityQueryInputBase.

Наследует все поля от ReachabilityQueryInput.

Проверки:

• Должен быть указан по крайней мере один из source_property_value или target_property_value

Пример:

# Preferred: keyword arguments
result = graph.k_hop(
    source_property_value="user123",
    max_hop_count=3,
    participating_edge_labels=["accessed"]
)

# DEPRECATED — will be removed in a future version. Use keyword arguments above.
from sentinel_graph.builders.query_input import K_HopQueryInput
result = graph.k_hop(query_input=K_HopQueryInput(source_property_value="user123"))

BlastRadiusQueryInput

Входные параметры для запроса радиуса взрыва от источника к целевым узлам. Наследует от ReachabilityQueryInputBase.

Наследует все поля от ReachabilityQueryInput, со следующими обязательными полями:

Проверки:

• source_property_value является обязательным
• target_property_value является обязательным

Пример:

# Preferred: keyword arguments
result = graph.blast_radius(
    source_property_value="user123",
    target_property_value="device456",
    participating_edge_labels=["accessed", "connected_to"]
)

# DEPRECATED — will be removed in a future version. Use keyword arguments above.
from sentinel_graph.builders.query_input import BlastRadiusQueryInput
result = graph.blast_radius(query_input=BlastRadiusQueryInput(
    source_property_value="user123",
    target_property_value="device456"
))

CentralityQueryInput

Входные параметры для запроса анализа централизации. Наследует от QueryInputBase.

Перечисление CentralityType

Fields

Пример:

# Preferred: keyword arguments (works for all centrality types)
result = graph.centrality(
    centrality_type=CentralityType.Edge,  # or CentralityType.Node (default)
    participating_edge_labels=["accessed", "connected_to"],
    threshold=5,
    max_results=100
)

# DEPRECATED — will be removed in a future version. Use keyword arguments above.
from sentinel_graph.builders.query_input import CentralityQueryInput, CentralityType
result = graph.centrality(query_input=CentralityQueryInput(
    centrality_type=CentralityType.Edge,
    participating_edge_labels=["accessed"]
))

RankedQueryInput

Входные параметры для ранжированных аналитических запросов. Наследует от QueryInputBase.

Fields

Пример:

# Preferred: keyword arguments
result = graph.ranked(
    rank_property_name="risk_score",
    threshold=5,
    decay_factor=2,
    max_results=50
)

# DEPRECATED — will be removed in a future version. Use keyword arguments above.
from sentinel_graph.builders.query_input import RankedQueryInput
result = graph.ranked(query_input=RankedQueryInput(
    rank_property_name="risk_score",
    threshold=5
))

Результаты запроса

QueryResult

Результат запроса графа с отложенным доступом к кадру данных.

constructor

QueryResult(raw_response: Dict[str, Any], graph: Graph)

Параметры:

• raw_response (Dict[str, Any]): словарь необработанных ответов API
• graph (Graph): ссылка на родительский граф

Примечание: Обычно создается с помощью Graph.query(), а не создает экземпляр напрямую.

Методы

to_dataframe

def to_dataframe() -> DataFrame

Преобразует результат запроса в кадр данных Spark.

Возвращает:

• DataFrame: результат запроса в виде кадра данных Spark

Поднимает:

• ValueError: если преобразование завершается ошибкой

Пример:

result = graph.query("MATCH (u:user) RETURN u")
df = result.to_dataframe()
df.show()

get_raw_data

def get_raw_data() -> Dict[str, Any]

Получение раздела RawData из ответа.

Возвращает:

• Dict[str, Any]: словарь с необработанными метаданными или пустым диктом, если он отсутствует

Пример:

result = graph.query("MATCH (u:user) RETURN u")
metadata = result.get_raw_data()

show

def show(format: str = "visual") -> None

Отображение результатов запроса в различных форматах.

Параметры:

• format (str, default="visual"): формат вывода
  • "table": полные таблицы dataframe (все столбцы)
  • "visual": интерактивная визуализация графа с подключаемым модулем VSC
  • "all": отображение всех форматов

Поднимает:

• ValueError: если формат не является одним из поддерживаемых значений.

Пример:

result = graph.query("MATCH (u:user)-[r:accessed]->(d:device) RETURN u, r, d")
result.show()  # Visual by default
result.show(format="table")  # Table format

Полный пример (рекомендуется — версия 0.3+)

# 0. Imports
from sentinel_graph import GraphSpecBuilder, Graph

# 1. Define graph specification
spec = (
    GraphSpecBuilder.start()
    
    .add_node("User")
        .from_dataframe(user_nodes)  # native Spark DF from groupBy → no .df
            .with_columns(
                "UserId", "UserDisplayName", "UserPrincipalName",
                "DistinctLocationCount", "DistinctIPCount", "DistinctAppCount",
                "TotalSignIns", "RiskySignInCount", "ImpossibleTravelFlag",
                key="UserId", display="UserDisplayName"
            )
    
    .add_node("IPAddress")
        .from_dataframe(ip_nodes)  # native Spark DF from groupBy → no .df
            .with_columns(
                "IPAddress", "UniqueUsers", "UniqueLocations",
                "SignInCount", "RiskySignInCount", "SharedIPFlag",
                key="IPAddress", display="IPAddress"
            )
    
    .add_edge("UsedIP")
        .from_dataframe(edge_used_ip)  # native Spark DF → no .df
            .source(id_column="UserId", node_type="User")
            .target(id_column="IPAddress", node_type="IPAddress")
            .with_columns(
                "SignInCount", "FirstSeen", "LastSeen", "EdgeKey",
                key="EdgeKey", display="EdgeKey"
            )
   
    .done()
)

# 2. Inspect schema before building (GraphSpec owns this)
spec.show_schema()

# 3. Build: prepares data + publishes graph → returns Graph
graph = Graph.build(spec)
print(f"Build status: {graph.build_status.status}")

# 4. Query the graph (query lives on Graph)
result = graph.query("MATCH (u:user)-[used:UsedIP]->(ip:IPAddress) RETURN * LIMIT 100")
result.show()

# 5. Access data via delegation
df = result.to_dataframe()
df.printSchema()

# 6. Graph algorithms
gf = graph.to_graphframe()
pagerank_result = gf.pageRank(resetProbability=0.15, maxIter=10)
pagerank_result.vertices.select("id", "pagerank").show()

# 7. Fetch an existing graph (no spec needed)
graph = Graph.get("my_existing_graph", context=context)
graph.query("MATCH (n) RETURN n LIMIT 10").show()

Примечания по шаблонам конструктора

Fluent API

Все построители поддерживают цепочку методов для читаемых декларативных определений графов:

builder.add_node("user") \
    .from_table("Users") \
    .with_columns("id", "name", key="id", display="name") \
    .add_edge("follows")

Схемы объединения

Несколько ребер с одинаковым псевдонимом автоматически объединяются с объединенными свойствами:

# Both edges use alias "sign_in" - they will be merged into one schema edge
builder.add_edge("sign_in") \
    .from_table("AzureSignins") \
    .source(id_column="UserId", node_type="AZuser") \
    .target(id_column="DeviceId", node_type="device")

builder.add_edge("sign_in") \
    .from_table("EntraSignins") \
    .source(id_column="UserId", node_type="EntraUser") \
    .target(id_column="DeviceId", node_type="device")

Автоматическая настройка

Многие поля имеют разумные значения по умолчанию:

• Метки узла и края по умолчанию — их псевдонимы
• Свойства автоматически выводятся из исходных схем
• Группы сущностей по умолчанию для первичных меток и типов связей

Отложенная оценка

Кадры данных и ресурсы загружаются отложенно и кэшируются:

• graph_spec.nodes и graph_spec.edges загружаются при первом доступе
• Результаты запроса создают кадры данных только при запросе

Связанные материалы

• Обзор пользовательских графов
• Создание пользовательских графов