Dokumentacja interfejsu API programu Graph Builder (wersja zapoznawcza)

Klasa sentinel_graph umożliwia interakcję z wykresem Microsoft Sentinel, umożliwiając definiowanie schematu grafu, przekształcanie danych z usługi Microsoft Sentinel data lake w węzły i krawędzie, publikowanie grafu, graf zapytań i uruchamianie zaawansowanych algorytmów grafów. Ta klasa jest przeznaczona do pracy z sesjami platformy Spark w notesach Jupyter działających na Microsoft Sentinel obliczeniach platformy Spark.

GraphSpecBuilder

Klasa GraphSpecBuilder zapewnia płynnego konstruktora do tworzenia specyfikacji grafu z potokami danych i integracją schematów.

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

# Recommended
from sentinel_graph import GraphSpecBuilder

Konstruktor

GraphSpecBuilder(context: ExecutionContext)

Parametry:

• context (ExecutionContext): Kontekst wykonywania zawierający sesję i konfigurację platformy Spark

Podnosi:

• ValueError: Jeśli kontekst ma wartość Brak lub nie można określić nazwy grafu

Metody statyczne

start

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

Zdefiniuj nowego płynnego konstruktora wykresów.

Parametry:

• context (ExecutionContext, opcjonalnie): Wystąpienie ExecutionContext. Jeśli nie, używa kontekstu domyślnego.

Zwraca:

• GraphSpecBuilder: Nowe wystąpienie konstruktora

Przykład:

builder = GraphSpecBuilder.start(context=context)

Metody wystąpień

add_node

def add_node(alias: str) -> NodeBuilderInitial

Rozpocznij tworzenie definicji węzła.

Parametry:

• alias (str): Unikatowy identyfikator tego węzła w grafie

Zwraca:

• NodeBuilderInitial: Konstruktor węzłów w stanie początkowym

Przykład:

builder.add_node("user")

add_edge

def add_edge(alias: str) -> EdgeBuilderInitial

Rozpocznij tworzenie definicji krawędzi.

Parametry:

• alias (str): Identyfikator tej krawędzi w grafie (może być współużytkowany na wielu krawędziach)

Zwraca:

• EdgeBuilderInitial: Konstruktor krawędzi w stanie początkowym

Przykład:

builder.add_edge("accessed")

done

def done() -> GraphSpec

Finalizowanie specyfikacji grafu i zwracanie wystąpienia GraphSpec.

Zwraca:

• GraphSpec: Ukończ specyfikację grafu za pomocą potoku danych i schematu

Podnosi:

• ValueError: Jeśli wykres nie ma węzłów ani krawędzi lub jeśli weryfikacja zakończy się niepowodzeniem

Przykład:

graph_spec = builder.done()

GraphSpec

Specyfikacja grafu z możliwościami potoku danych, schematu i wyświetlania.

Konstruktor

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

Parametry:

• name (str): Nazwa grafu
• context (ExecutionContext): Kontekst wykonywania
• graph_schema (GraphSchema): definicja schematu grafu
• etl_pipeline (ETLPipeline, opcjonalnie): potok danych na potrzeby przygotowywania grafu

Właściwości

nodes

def nodes() -> DataFrame

Pobierz element DataFrame węzłów (z opóźnieniem, buforowany). Automatycznie określa źródło na podstawie potoku danych lub tabeli lake.

Zwraca:

• DataFrame: Ramka danych platformy Spark zawierająca wszystkie węzły

Podnosi:

• ValueError: Jeśli brakuje kontekstu lub nie można załadować ramek danych

edges

def edges() -> DataFrame

Pobierz ramkę danych krawędzi (z opóźnieniem, buforowane). Automatycznie określa źródło na podstawie potoku danych lub tabeli lake.

Zwraca:

• DataFrame: Ramka danych platformy Spark zawierająca wszystkie krawędzie

Podnosi:

• ValueError: Jeśli brakuje kontekstu lub nie można załadować ramek danych

Metody

build_graph_with_data

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

Wykonaj potok danych i opublikuj graf. Wewnętrznie wywołuje Graph.build(self)element , schowa zwrócony Graphsłownik i zwraca słownik zgodny z poprzednimi wersjami.

Zwraca:

• Dict[str, Any]: Słownik zawierający:
  • etl_result: Wyniki przygotowywania danych
  • api_result: Publikowanie wyników (jeśli się powiedzie)
  • api_error: Ciąg błędu (jeśli publikowanie nie powiodło się)
  • instance_name: Nazwa wystąpienia grafu
  • status: "published" lub "prepared"

Przykład:

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

get_schema

def get_schema() -> GraphSchema

Pobierz schemat grafu.

Zwraca:

• GraphSchema: Definicja schematu grafu

get_pipeline

def get_pipeline() -> Optional[ETLPipeline]

Pobierz potok danych (Brak dla istniejących grafów).

Zwraca:

• ETLPipeline lub None: Potok danych, jeśli jest dostępny

to_graphframe

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

Konwertuj cały graf na element GraphFrame na potrzeby uruchamiania algorytmów grafów. Działa tylko na danych lokalnych (z potoku danych lub tabeli lake).

Parametry:

• column_mapping (Dict[str, str], opcjonalnie): Niestandardowe mapowanie kolumn za pomocą kluczy:
  • "id": Nazwa kolumny identyfikatora wierzchołka
  • "source_id": Nazwa kolumny identyfikatora źródła krawędzi
  • "target_id": Nazwa kolumny docelowego identyfikatora krawędzi

Zwraca:

• GraphFrame: obiekt GraphFrame ze wszystkimi wierzchołkami i krawędziami

Podnosi:

• ValueError: Jeśli funkcja ExecutionContext nie jest dostępna

Przykład:

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

show

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

Wyświetlanie danych grafu w różnych formatach.

Parametry:

• limit (int, default=100): maksymalna liczba węzłów/krawędzi do wyświetlenia
• viz_format (str, default="visual"): format danych wyjściowych
  • "table": Pełne tabele ramek danych (wszystkie kolumny)
  • "visual": Interaktywna wizualizacja grafu
  • "all": Pokaż wszystkie formaty

Podnosi:

• ValueError: Jeśli format nie jest jedną z obsługiwanych wartości

Przykład:

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

show_schema

def show_schema() -> None

Wyświetlanie schematu grafu jako interaktywnej wizualizacji grafu.

Przykład:

spec.show_schema()

Graph

Wystąpienie grafu z możliwością wykonywania zapytań. Utworzone za pośrednictwem Graph.get() (istniejącego grafu) lub Graph.build() (na podstawie GraphSpecelementu ).

Konstruktor

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

Parametry:

• name (str): Nazwa grafu
• context (ExecutionContext): Kontekst wykonywania
• spec (GraphSpec, opcjonalnie): Dołączona specyfikacja grafu (ustawiona przez Graph.build())
• build_status (BuildStatus, opcjonalnie): Metadane wyników kompilacji (ustawione przez Graph.build())

Podnosi:

• ValueError: Jeśli parametr ExecutionContext to Brak

Metody statyczne

get

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

Pobieranie wystąpienia grafu z istniejącego grafu. Zwrócone elementy Graph mają spec=None wartości i build_status=None.

Parametry:

• name (str): Nazwa wystąpienia grafu
• context (ExecutionContext, opcjonalnie): Kontekst wykonywania (domyślnie ExecutionContext.default())

Zwraca:

• Graph: Wystąpienie programu Graph

Podnosi:

• ValueError: Jeśli nazwa grafu jest pusta lub wystąpienie grafu nie istnieje

Przykład:

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

prepare

Graph.prepare(spec: GraphSpec) -> Graph

Uruchom etap przygotowywania danych dla GraphSpec elementu bez publikowania. Użyj później publish() , aby zarejestrować graf i umożliwić wykonywanie zapytań.

Parametry:

• spec (GraphSpec): specyfikacja grafu do przygotowania

Zwraca:

• Graph: wystąpienie grafu z spec dołączonymi i build_status.status == "prepared"

Podnosi:

• ValueError: jeśli specyfikacja nie ma potoku danych lub nie ma kontekstu wykonywania
• RuntimeError: Jeśli wykonanie potoku danych zakończy się niepowodzeniem

Przykład:

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

Skompiluj wykres na podstawie elementu , GraphSpec przygotowując dane i publikując. Wewnętrznie wywołuje, Graph.prepare(spec) a następnie próbuje graph.publish(). W przeciwieństwie do wywoływania tych dwóch metod oddzielnie, błędy publikowania są przechwytywany — zwracany graf ma build_status.status == "prepared" i build_status.api_error ustawia zamiast podnoszenia.

Parametry:

• spec (GraphSpec): specyfikacja grafu do skompilowania na podstawie

Zwraca:

• Graph: Wystąpienie grafu z spec dołączonym i build_status wypełnionym

Podnosi:

• ValueError: jeśli specyfikacja nie ma potoku danych lub nie ma kontekstu wykonywania
• RuntimeError: Jeśli wykonanie potoku danych zakończy się niepowodzeniem

Przykład:

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")

Właściwości

nodes

def nodes() -> Optional[DataFrame]

Pobierz element DataFrame węzłów. Deleguje do self.spec.nodes , gdy specyfikacja jest dołączona; zwraca None inaczej.

edges

def edges() -> Optional[DataFrame]

Pobierz ramkę danych krawędzi. Deleguje do self.spec.edges , gdy specyfikacja jest dołączona; zwraca None inaczej.

schema

def schema() -> Optional[GraphSchema]

Pobierz schemat grafu. Deleguje do self.spec.get_schema() , gdy specyfikacja jest dołączona; zwraca None inaczej.

Metody

query

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

Wykonaj zapytanie względem wystąpienia grafu przy użyciu protokołu GQL.

Parametry:

• query_string (str): ciąg zapytania grafu (język GQL)
• query_language (str, default="GQL"): język zapytań

Zwraca:

• QueryResult: obiekt zawierający węzły, krawędzie i metadane

Podnosi:

• ValueError: Jeśli brakuje sesji ExecutionContext lub Spark
• RuntimeError: Jeśli inicjowanie klienta lub wykonanie zapytania zakończy się niepowodzeniem

Przykład:

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

[! UWAGA] reachability(query_input=ReachabilityQueryInput(...)) jest nadal akceptowana, ale emituje DeprecationWarning i zostanie usunięta w przyszłej wersji.

Wykonaj analizę dostępności między węzłami źródłowym i docelowym.

Parametry:

• source_property_value (str): Wartość zgodna z właściwością źródłową (zweryfikowana w czasie wykonywania. Należy podano, jeśli nie używasz programu query_input)
• target_property_value (str): Wartość zgodna z właściwością docelową (zweryfikowana w czasie wykonywania. Należy podano, jeśli nie używasz programu query_input)
• source_property (Opcjonalnie[str]): nazwa właściwości do filtrowania węzłów źródłowych
• participating_source_node_labels (Opcjonalnie[Lista[str]]): Etykiety węzłów do rozważenia jako źródła
• target_property (Opcjonalnie[str]): nazwa właściwości do filtrowania węzłów docelowych
• participating_target_node_labels (Opcjonalnie[Lista[str]]): etykiety węzłów do rozważenia jako elementy docelowe
• participating_edge_labels (Opcjonalnie[Lista[str]]): Etykiety krawędzi do przejścia
• is_directional (bool): Czy krawędzie są kierunkowe (domyślnie: True)
• min_hop_count (int): minimalne przeskoki (domyślnie: 1)
• max_hop_count (int): maksymalna liczba przeskoków (wartość domyślna: 4)
• shortest_path (bool): Zwracaj tylko najkrótsze ścieżki (domyślnie: False)
• max_results (int): maksymalna liczba wyników (wartość domyślna: 500)

Podnosi:

• ValueError: Jeśli source_property_value brakuje lubtarget_property_value, , min_hop_count < 1max_hop_count < min_hop_count, lubmax_results < 1
• RuntimeError: Jeśli inicjowanie klienta lub wykonanie zapytania zakończy się niepowodzeniem

Zwraca:

• QueryResult: zawiera ścieżki osiągalności

Przykład:

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

Wykonaj analizę k-hopu z danego węzła źródłowego.

Parametry:

• Tak samo jak reachability

Sprawdzania poprawności:

• Co najmniej jeden z source_property_value lub target_property_value musi zostać dostarczony

Podnosi:

• ValueError: Jeśli nie podano ani nie source_property_valuetarget_property_value podano ograniczeń liczbowych (tak samo jak reachability)
• RuntimeError: Jeśli inicjowanie klienta lub wykonanie zapytania zakończy się niepowodzeniem

Zwraca:

• QueryResult: Zawiera wyniki k-hopu

Przykład:

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

Wykonaj analizę promienia wybuchu z węzła źródłowego do węzła docelowego.

Parametry:

• source_property_value (str): Wartość identyfikująca węzeł źródłowy (zweryfikowana w czasie wykonywania. Należy podano, jeśli nie używasz programu query_input)
• target_property_value (str): Wartość identyfikująca węzeł docelowy (zweryfikowana w czasie wykonywania. Należy podano, jeśli nie używasz programu query_input)
• Inne parametry: takie same jak reachability

Podnosi:

• ValueError: jeśli source_property_value brakuje lub target_property_value lub jeśli naruszono ograniczenia liczbowe (takie same jak reachability)
• RuntimeError: Jeśli inicjowanie klienta lub wykonanie zapytania zakończy się niepowodzeniem

Zwraca:

• QueryResult: Zawierające wyniki promienia wybuchu

Przykład:

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

Przeprowadzanie analizy centralności na wykresie.

Parametry:

• participating_source_node_labels (Opcjonalnie[Lista[str]]): Etykiety węzłów źródłowych
• participating_target_node_labels (Opcjonalnie[Lista[str]]): Etykiety węzłów docelowych
• participating_edge_labels (Opcjonalnie[Lista[str]]): Etykiety krawędzi do przejścia
• threshold (int): Minimalna ocena centralności (domyślna: 3); musi być nieujemna
• centrality_type (CentralityType): CentralityType.Node lub CentralityType.Edge (domyślnie: None, wraca do CentralityType.Node)
• max_paths (int): Maksymalna liczba ścieżek do rozważenia (wartość domyślna: 1000000; 0 = wszystkie ścieżki); musi być nieujemna
• is_directional (bool): Czy krawędzie są kierunkowe (domyślnie: True)
• min_hop_count (int): Minimalne przeskoki (domyślne: 1); muszą być ≥ 1
• max_hop_count (int): maksymalna liczba przeskoków (wartość domyślna: 4); musi być ≥ min_hop_count
• shortest_path (bool): Zwracaj tylko najkrótsze ścieżki (domyślnie: False)
• max_results (int): Maksymalna liczba wyników (wartość domyślna: 500); musi być ≥ 1

Podnosi:

• ValueError: Jeśli threshold < 0, max_paths < 0, min_hop_count < 1, , max_hop_count < min_hop_countlub max_results < 1
• RuntimeError: Jeśli inicjowanie klienta lub wykonanie zapytania zakończy się niepowodzeniem

Zwraca:

• QueryResult: Zawierające metryki centralności

Przykład:

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

Przeprowadzanie analizy rankingowej na wykresie.

Parametry:

• rank_property_name (str): Nazwa właściwości do użycia na potrzeby klasyfikacji (zweryfikowana w czasie wykonywania. Należy podano, jeśli nie używasz programu query_input)
• threshold (int): Tylko ścieżki zwracane powyżej tej wagi (domyślnie: 0); muszą być nieujemne
• max_paths (int): Maksymalna liczba ścieżek do rozważenia (wartość domyślna: 1000000; 0 = wszystkie ścieżki); musi być nieujemna
• decay_factor (zmiennoprzecinkowa): klasyfikacja rozkładu na krok; 2 oznacza zmniejszenie o połowę (wartość domyślna: 1); musi być nieujemna
• is_directional (bool): Czy krawędzie są kierunkowe (domyślnie: True)
• min_hop_count (int): Minimalne przeskoki (domyślne: 1); muszą być ≥ 1
• max_hop_count (int): maksymalna liczba przeskoków (wartość domyślna: 4); musi być ≥ min_hop_count
• shortest_path (bool): Zwracaj tylko najkrótsze ścieżki (domyślnie: False)
• max_results (int): Maksymalna liczba wyników (wartość domyślna: 500); musi być ≥ 1

Podnosi:

• ValueError: Jeśli rank_property_name brakuje, , threshold < 0, max_paths < 0decay_factor < 0, min_hop_count < 1, , max_hop_count < min_hop_count, lubmax_results < 1
• RuntimeError: Jeśli inicjowanie klienta lub wykonanie zapytania zakończy się niepowodzeniem

Zwraca:

• QueryResult: Zawiera węzły/krawędzie sklasyfikowane

Przykład:

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

Konwertuj cały graf na element GraphFrame. Używa danych specyfikacji, jeśli są dostępne; odczyty z tabel lake w przeciwnym razie.

Parametry:

• column_mapping (Dict[str, str], opcjonalnie): Mapowanie kolumn niestandardowych

Zwraca:

• GraphFrame: obiekt GraphFrame ze wszystkimi wierzchołkami i krawędziami

Przykład:

gf = graph.to_graphframe()

show

def show() -> None

Wyświetl informacje o grafie. Deleguje do spec.show() wyświetlania rozbudowanego, gdy specyfikacja jest dołączona; w przeciwnym razie wyświetla minimalne informacje.

show_schema

def show_schema() -> None

Wyświetlanie schematu grafu. Deleguje do spec.show_schema() , gdy specyfikacja jest dołączona; wyświetla komunikat wskazujący, że żaden schemat nie jest dostępny w przeciwnym razie.

publish (nowość w wersji 0.3.3)

def publish() -> Graph

Zarejestruj wykres przy użyciu interfejsu API, dzięki czemu można wykonywać zapytania. Wywołaj tę metodę po Graph.prepare() (lub na dowolnym Graph , który ma dołączoną specyfikację), aby opublikować wystąpienie grafu.

Zwraca:

• Graph: Samodzielne tworzenie łańcuchów metod

Podnosi:

• ValueError: Jeśli nie jest dołączona żadna specyfikacja lub brakuje kontekstu
• RuntimeError: Jeśli publikowanie zakończy się niepowodzeniem

Przykład:

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

BuildStatus

Klasa danych z metadanymi z Graph.build() operacji.

Pola

Ścieżki konstrukcyjne

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)

Przykład:

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")

Konstruktorzy węzłów

NodeBuilderInitial

Stan początkowy konstruktora węzłów: dostępne są tylko metody źródła danych.

Konstruktor

NodeBuilderInitial(alias: str, graph_builder: GraphSpecBuilder)

Uwaga: Zazwyczaj tworzone za pośrednictwem GraphSpecBuilder.add_node()programu , nie są tworzone bezpośrednio.

Metody

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

Ustaw tabelę jako źródło danych z inteligentną rozdzielczością bazy danych.

Parametry:

• table_name (str): Nazwa tabeli (wymagana)
• database (str, opcjonalnie): jawna nazwa bazy danych (ma pierwszeństwo przed domyślnym kontekstem)
• time_generated_start (str, optional) i time_generated_end (str, optional): Opcjonalne pola sygnatury czasowej, gdy są używane razem aktywuje zoptymalizowane odczyty różnicowe. Zalecamy użycie formy kanonicznej: "rrrr-MM-dd HH:mm:ss".

Zwraca:

• NodeBuilderSourceSet: Konstruktor do dalszej konfiguracji

Podnosi:

• ValueError: Jeśli nie znaleziono tabeli lub znaleziono wiele tabel powodujących konflikt

Kolejność rozpoznawania bazy danych:

1. Parametr jawny database (najwyższy priorytet)
2. ExecutionContext.default_database
3. Przeszukaj wszystkie bazy danych (z wykrywaniem konfliktów)

Przykład:

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

Ustaw element Spark DataFrame jako źródło danych.

Parametry:

• dataframe (DataFrame): Ramka danych platformy Spark

Zwraca:

• NodeBuilderSourceSet: Konstruktor do dalszej konfiguracji

Przykład:

# 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

Konstruktor węzłów po ustawieniu źródła danych: dostępne metody konfiguracji.

Konstruktor

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

Uwaga: Utworzone wewnętrznie przez metody źródłowe NodeBuilderInitial.

Metody

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

Zastosuj filtrowanie zakresu czasu do źródła danych węzła.

Parametry:

• time_column (str): Nazwa kolumny zawierająca dane sygnatury czasowej (wymagane)
• start_time (str lub datetime, opcjonalnie): data rozpoczęcia ("10/20/25", "2025-10-20" lub obiekt datetime)
• end_time (str lub datetime, opcjonalnie): data zakończenia (takie same formaty jak start_time)
• lookback_hours (zmiennoprzecinkowa, opcjonalna): godziny do obejrzenia od teraz

Zwraca:

• NodeBuilderSourceSet: Samodzielne tworzenie łańcuchów metod

Podnosi:

• ValueError: Jeśli kolumna czasu nie została znaleziona w schemacie źródłowym

Logika zakresu czasu:

1. Jeśli podano start_time i end_time: użyj ich bezpośrednio
2. Jeśli podano tylko lookback_hours: end=now, start=now-lookback_hours
3. Jeśli nie podano żadnych informacji: brak filtrowania czasu
4. Jeśli początek/koniec I lookback_hours: początek/koniec mają pierwszeństwo

Przykład:

# 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

Ustaw etykietę węzła (domyślnie alias, jeśli nie jest wywoływany).

Parametry:

• label (str): Etykieta węzła

Zwraca:

• NodeBuilderSourceSet: Samodzielne tworzenie łańcuchów metod

Podnosi:

• ValueError: Jeśli etykieta została już ustawiona

Przykład:

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

with_columns

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

Skonfiguruj kolumny z wymaganym kluczem i oznaczeniem wyświetlania.

Parametry:

• *columns (str): Nazwy kolumn do uwzględnienia (co najmniej jeden wymagany)
• key (str): Nazwa kolumny, aby oznaczyć jako klucz (wymagane, musi być w kolumnach)
• display (str): Nazwa kolumny, aby oznaczyć jako wartość wyświetlaną (wymagane, musi być w kolumnach, może być taka sama jak klucz)

Zwraca:

• NodeBuilderSourceSet: Samodzielne tworzenie łańcuchów metod

Podnosi:

• ValueError: Jeśli weryfikacja zakończy się niepowodzeniem (zduplikowane kolumny, brak klucza/wyświetlacza itp.)

Uwagi:

• Właściwości są automatycznie kompilowane na podstawie typów kolumn

• Kolumna filtru czasu jest dodawana automatycznie, jeśli zostanie określona

• Typy właściwości są automatycznie wnioskowane ze schematu źródłowego

• Zobacz Ograniczenia

Przykład:

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

Zakończ ten węzeł i rozpocznij tworzenie innego węzła.

Parametry:

• alias (str): Alias nowego węzła

Zwraca:

• NodeBuilderInitial: Nowy konstruktor węzłów

Przykład:

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

Zakończ ten węzeł i rozpocznij tworzenie krawędzi.

Parametry:

• alias (str): alias krawędzi

Zwraca:

• EdgeBuilderInitial: Nowy konstruktor krawędzi

Przykład:

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

done

def done() -> GraphSpec

Zakończ ten węzeł i wypełnij specyfikację grafu.

Zwraca:

• GraphSpec: Kompletna specyfikacja grafu

Przykład:

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

Konstruktorzy krawędzi

EdgeBuilderInitial

Stan początkowy konstruktora krawędzi: dostępne są tylko metody źródła danych.

Konstruktor

EdgeBuilderInitial(alias: str, graph_builder: GraphSpecBuilder)

Uwaga: Zazwyczaj tworzone za pośrednictwem GraphSpecBuilder.add_edge()programu , nie są tworzone bezpośrednio.

Metody

from_table

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

Ustaw tabelę jako źródło danych z inteligentną rozdzielczością bazy danych.

Parametry:

• table_name (str): Nazwa tabeli (wymagana)
• database (str, opcjonalnie): jawna nazwa bazy danych

Zwraca:

• EdgeBuilderSourceSet: Konstruktor do dalszej konfiguracji

Podnosi:

• ValueError: Jeśli nie znaleziono tabeli lub znaleziono wiele tabel powodujących konflikt

Przykład:

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

from_dataframe

def from_dataframe(dataframe: DataFrame) -> EdgeBuilderSourceSet

Ustaw element Spark DataFrame jako źródło danych.

Parametry:

• dataframe (DataFrame): Ramka danych platformy Spark

Zwraca:

• EdgeBuilderSourceSet: Konstruktor do dalszej konfiguracji

Przykład:

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

EdgeBuilderSourceSet

Konstruktor edge po ustawieniu źródła danych: dostępne metody konfiguracji.

Konstruktor

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

Uwaga: Utworzone wewnętrznie przez metody źródłowe EdgeBuilderInitial.

Metody

with_label

def with_label(label: str) -> EdgeBuilderSourceSet

Ustaw typ/etykietę relacji krawędzi (domyślnie alias, jeśli nie jest wywoływany).

Parametry:

• label (str): Etykieta krawędzi

Zwraca:

• EdgeBuilderSourceSet: Samodzielne tworzenie łańcuchów metod

Podnosi:

• ValueError: Jeśli etykieta została już ustawiona

Przykład:

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

edge_label

def edge_label(label: str) -> EdgeBuilderSourceSet

Ustaw typ/etykietę relacji krawędzi (domyślnie alias, jeśli nie jest wywoływany).

Parametry:

• label (str): Etykieta krawędzi

Zwraca:

• EdgeBuilderSourceSet: Samodzielne tworzenie łańcuchów metod

Podnosi:

• ValueError: Jeśli etykieta została już ustawiona

Przykład:

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

source

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

Ustaw węzeł źródłowy z kolumną i etykietą identyfikatora.

Parametry:

• id_column (str): Nazwa kolumny zawierająca identyfikator węzła źródłowego
• node_type (str): Etykieta węzła źródłowego

Zwraca:

• EdgeBuilderSourceSet: Samodzielne tworzenie łańcuchów metod

Podnosi:

• ValueError: Jeśli źródło zostało już ustawione

Przykład:

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

Ustaw węzeł docelowy z kolumną i etykietą identyfikatora.

Parametry:

• id_column (str): Nazwa kolumny zawierająca identyfikator węzła docelowego
• node_type (str): Etykieta węzła docelowego

Zwraca:

• EdgeBuilderSourceSet: Samodzielne tworzenie łańcuchów metod

Podnosi:

• ValueError: Jeśli obiekt docelowy został już ustawiony

Przykład:

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

Zastosuj filtrowanie zakresu czasu do źródła danych krawędzi.

Parametry:

• time_column (str): Nazwa kolumny zawierająca dane sygnatury czasowej (wymagane)
• start_time (str lub datetime, opcjonalnie): data rozpoczęcia
• end_time (str lub datetime, opcjonalnie): data zakończenia
• lookback_hours (zmiennoprzecinkowa, opcjonalna): godziny do obejrzenia od teraz

Zwraca:

• EdgeBuilderSourceSet: Samodzielne tworzenie łańcuchów metod

Podnosi:

• ValueError: Jeśli kolumna czasu nie została znaleziona w schemacie źródłowym

Przykład:

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

Skonfiguruj kolumny z wymaganym kluczem i oznaczeniem wyświetlania.

Parametry:

• *columns (str): Nazwy kolumn do uwzględnienia (co najmniej jeden wymagany)
• key (str): Nazwa kolumny, aby oznaczyć jako klucz (wymagane, musi być w kolumnach)
• display (str): Nazwa kolumny, aby oznaczyć jako wartość wyświetlaną (wymagane, musi być w kolumnach)

Zwraca:

• EdgeBuilderSourceSet: Samodzielne tworzenie łańcuchów metod

Podnosi:

• ValueError: Jeśli weryfikacja nie powiedzie się

Przykład:

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

Zakończ tę krawędź i rozpocznij tworzenie węzła.

Parametry:

• alias (str): Alias nowego węzła

Zwraca:

• NodeBuilderInitial: Nowy konstruktor węzłów

add_edge

def add_edge(alias: str) -> EdgeBuilderInitial

Zakończ tę krawędź i zacznij tworzyć kolejną krawędź.

Parametry:

• alias (str): alias nowej krawędzi

Zwraca:

• EdgeBuilderInitial: Nowy konstruktor krawędzi

Przykład:

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

Sfinalizuj tę krawędź i wypełnij specyfikację grafu.

Zwraca:

• GraphSpec: Kompletna specyfikacja grafu

Klasy schematów

GraphDefinitionReference

Odwołanie do definicji grafu o nazwie i wersji.

Konstruktor

GraphDefinitionReference(
    fully_qualified_name: str,
    version: str
)

Parametry:

• fully_qualified_name (str): W pełni kwalifikowana nazwa grafu, do którego odwołuje się odwołanie
• version (str): Wersja grafu, do którego odwołuje się odwołanie

Podnosi:

• ValueError: Jeśli fully_qualified_name lub wersja jest pusta

Metody

to_dict

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

Serializuj do słownika.

Zwraca:

• Dict[str, Any]: Dokumentacja serializowana

Właściwość

Definicja właściwości z interfejsem bezpiecznym dla typu.

Konstruktor

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
)

Parametry:

• name (str): Nazwa właściwości
• property_type (PropertyType): Typ danych właściwości
• is_non_null (bool, default=False): czy właściwość jest wymagana
• description (str, default=""):Opis właściwości
• is_key (bool, default=False): czy właściwość jest kluczem
• is_display_value (bool, default=False): czy właściwość jest wartością wyświetlaną
• is_internal (bool, default=False): czy właściwość jest wewnętrzna

Podnosi:

• ValueError: Jeśli nazwa jest pusta lub weryfikacja kończy się niepowodzeniem

Metody klasy

key

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

Utwórz właściwość klucza z typowymi ustawieniami (is_key=True, is_display_value=True).

display

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

Utwórz właściwość wartości wyświetlanej (is_display_value=True).

Metody

describe

def describe(text: str) -> Property

Płynnie dodaj opis.

Parametry:

• text (str): Tekst opisu

Zwraca:

• Property: Samodzielne tworzenie łańcuchów metod

to_dict

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

Serializuj właściwość do słownika przy użyciu kluczy adnotacji z prefiksem @.

Zwraca:

• Dict[str, Any]: Właściwość serializowana

to_gql

def to_gql() -> str

Wygeneruj definicję właściwości GQL.

Zwraca:

• str: reprezentacja ciągu GQL

Węzeł edgenode

Odwołanie do węzła używane w definicjach krawędzi.

Konstruktor

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

Parametry:

• alias (str, opcjonalnie): alias węzła (automatycznie ustawiony na pierwszą etykietę, jeśli brak lub pusty)
• labels (List[str]): Etykiety węzłów (co najmniej jedna wymagana)

Podnosi:

• ValueError: Jeśli lista etykiet jest pusta
• TypeError: Jeśli etykiety nie są ciągami

Auto-mutacja:

• Jeśli alias ma wartość Brak lub jest pusty, jest ustawiony na pierwszą etykietę

Metody

to_dict

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

Serializuj do słownika.

Zwraca:

• Dict[str, Any]: Dokumentacja zserializowanego węzła krawędzi

Node

Definicja węzła z interfejsem bezpiecznym dla typu.

Konstruktor

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
)

Parametry:

• alias (str, default=""): alias węzła (automatycznie ustawiany na pierwszą etykietę, jeśli jest pusta)
• labels (List[str]): Etykiety węzłów (co najmniej jedna wymagana)
• implies_labels (List[str], default=[]): Etykiety dorozumiane
• properties (List[Property], default=[]): Właściwości węzła
• description (str, default=""): opis węzła
• entity_group (str, default=""): nazwa grupy jednostek
• dynamic_labels (bool, default=False): czy węzeł ma etykiety dynamiczne
• abstract_edge_aliases (bool, default=False): czy węzeł używa abstrakcyjnych aliasów krawędzi

Podnosi:

• ValueError: Jeśli weryfikacja zakończy się niepowodzeniem (brak etykiet, bez właściwości klucza, bez właściwości wyświetlania itp.)

Auto-mutacja:

• Jeśli alias jest pusty, jest ustawiony na pierwszą etykietę
• Jeśli entity_group jest pusta, jest ustawiona na etykietę podstawową

Metody

get_primary_label

def get_primary_label() -> Optional[str]

Pobierz etykietę podstawową (pierwszą).

Zwraca:

• str lub None: Etykieta podstawowa lub Brak, jeśli nie ma etykiet

get_entity_group_name

def get_entity_group_name() -> str

Pobierz nazwę grupy jednostek lub powrót do etykiety podstawowej.

Zwraca:

• str: Nazwa grupy jednostek

get_primary_key_property_name

def get_primary_key_property_name() -> Optional[str]

Pobierz nazwę właściwości klucza podstawowego.

Zwraca:

• str lub None: Nazwa właściwości klucza podstawowego

get_properties

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

Uzyskiwanie właściwości jako słownika w celu ułatwienia dostępu.

Zwraca:

• Dict[str, Property]: Właściwości z kluczem według nazwy

get_property

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

Pobierz określoną właściwość według nazwy.

Parametry:

• name (str): Nazwa właściwości

Zwraca:

• Property lub None: Właściwość, jeśli znaleziono

add_property

def add_property(prop: Property) -> None

Dodaj właściwość do tego węzła.

Parametry:

• prop (Właściwość): właściwość do dodania

Podnosi:

• ValueError: Jeśli nazwa właściwości jest zduplikowana

is_dynamically_labeled

def is_dynamically_labeled() -> bool

Sprawdź, czy węzeł ma etykiety dynamiczne.

Zwraca:

• bool: Prawda, jeśli włączono etykiety dynamiczne

is_abstract_edge_node_aliases

def is_abstract_edge_node_aliases() -> bool

Sprawdź, czy węzeł używa abstrakcyjnych aliasów węzłów krawędzi.

Zwraca:

• bool: Prawda, jeśli włączono abstrakcyjne aliasy krawędzi

describe

def describe(text: str) -> Node

Płynnie dodaj opis.

Parametry:

• text (str): Tekst opisu

Zwraca:

• Node: Samodzielne tworzenie łańcuchów metod

to_dict

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

Serializowanie węzła do słownika.

Zwraca:

• Dict[str, Any]: Węzeł serializowany

to_gql

def to_gql() -> str

Generowanie definicji węzła GQL.

Zwraca:

• str: reprezentacja ciągu GQL

Podnosi:

• ValueError: Jeśli w węźle nie ma wymaganych pól dla protokołu GQL

Metody klasy

create

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

Utwórz węzeł ze wszystkimi wymaganymi polami.

Parametry:

• alias (str): Alias węzła
• labels (List[str]): Etykiety węzłów
• properties (List[Property]): Właściwości węzła
• description (str, default=""): opis węzła
• entity_group (str, default=""): nazwa grupy jednostek

Zwraca:

• Node: Nowe wystąpienie węzła

Edge

Definicja krawędzi z interfejsem bezpiecznym dla typów.

Konstruktor

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
)

Parametry:

• relationship_type (str): Typ relacji krawędzi (na przykład "FOLLOWS", "OWNS")
• source_node_label (str): Etykieta węzła źródłowego
• target_node_label (str): Etykieta węzła docelowego
• direction (EdgeDirection, default=DIRECTED_RIGHT): Kierunek krawędzi
• properties (List[Property], default=[]): Właściwości krawędzi
• description (str, default=""): opis krawędzi
• entity_group (str, default=""): nazwa grupy jednostek
• dynamic_type (bool, default=False): czy krawędź ma typ dynamiczny

Podnosi:

• ValueError: Jeśli weryfikacja nie powiedzie się

Auto-mutacja:

• labels lista jest wypełniana automatycznie za pomocą polecenia [relationship_type]
• Jeśli entity_group jest pusta, jest ustawiona na relationship_type

Właściwości

edge_type

def edge_type() -> str

Alias zgodności z poprzednimi wersjami dla relationship_type.

Zwraca:

• str: Typ relacji

Metody

get_entity_group_name

def get_entity_group_name() -> str

Pobierz nazwę grupy jednostek lub powrót do typu relacji.

Zwraca:

• str: Nazwa grupy jednostek

is_dynamic_type

def is_dynamic_type() -> bool

Sprawdź, czy krawędź ma typ dynamiczny.

Zwraca:

• bool: Prawda, jeśli typ dynamiczny

add_property

def add_property(edge_property: Property) -> None

Dodaj właściwość do tej krawędzi.

Parametry:

• edge_property (Właściwość): właściwość do dodania

describe

def describe(text: str) -> Edge

Płynnie dodaj opis.

Parametry:

• text (str): Tekst opisu

Zwraca:

• Edge: Samodzielne tworzenie łańcuchów metod

to_dict

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

Serializowanie krawędzi do słownika.

Zwraca:

• Dict[str, Any]: Serializowana krawędź

to_gql

def to_gql() -> str

Generowanie definicji krawędzi GQL.

Zwraca:

• str: reprezentacja ciągu GQL

Metody klasy

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

Utwórz krawędź ze wszystkimi wymaganymi polami.

Parametry:

• relationship_type (str): Typ relacji krawędzi
• source_node_label (str): Etykieta węzła źródłowego
• target_node_label (str): Etykieta węzła docelowego
• properties (List[Property], opcjonalnie): właściwości krawędzi
• description (str, default=""): opis krawędzi
• entity_group (str, default=""): nazwa grupy jednostek

Zwraca:

• Edge: Nowe wystąpienie krawędzi

GraphSchema

Definicja schematu grafu z interfejsem bezpiecznym dla typu.

Konstruktor

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 = ""
)

Parametry:

• name (str): nazwa schematu grafu
• nodes (List[Node], default=[]): Definicje węzłów
• edges (List[Edge], default=[]): Definicje krawędzi
• base_graphs (List[GraphSchema], default=[]): Schematy wykresów bazowych
• description (str, default=""): Opis schematu
• version (str, default="1.0"): wersja schematu
• fully_qualified_name (str, default=""): W pełni kwalifikowana nazwa
• namespace (str, default=""): Przestrzeń nazw

Podnosi:

• ValueError: Jeśli weryfikacja zakończy się niepowodzeniem (zduplikowane aliasy, krawędzie odwołują się do nieistniejących węzłów itp.)

Metody

get_fully_qualified_name

def get_fully_qualified_name() -> str

Pobierz w pełni kwalifikowaną nazwę.

Zwraca:

• str: W pełni kwalifikowana nazwa

get_namespace

def get_namespace() -> str

Pobierz przestrzeń nazw z w pełni kwalifikowanej nazwy lub zwróć wartość domyślną.

Zwraca:

• str:Obszaru nazw

get_version

def get_version() -> str

Pobierz wersję.

Zwraca:

• str: Ciąg wersji

get_node

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

Pobierz węzeł według etykiety lub aliasu.

Parametry:

• label_or_alias (str): Etykieta węzła lub alias

Zwraca:

• Node lub None: Węzeł w przypadku znalezienia

get_edge

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

Uzyskiwanie krawędzi według nazwy/typu.

Parametry:

• name (str): Typ relacji krawędzi

Zwraca:

• Edge lub None: Przeglądarka Edge, jeśli została znaleziona

add_node

def add_node(node: Node) -> None

Dodaj węzeł do tego wykresu.

Parametry:

• node (Węzeł): węzeł do dodania

Podnosi:

• ValueError: Jeśli alias węzła jest zduplikowany

add_edge

def add_edge(edge: Edge) -> None

Dodaj krawędź do tego wykresu.

Parametry:

• edge (Przeglądarka Brzegowa): przeglądarka Edge do dodania

Podnosi:

• ValueError: Jeśli typ krawędzi jest zduplikowany

include_graph

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

Dodaj graf include (płynny interfejs API).

Parametry:

• fully_qualified_name (str): W pełni kwalifikowana nazwa grafu do uwzględnienia
• version (str): Wersja grafu do uwzględnienia

Zwraca:

• GraphSchema: Samodzielne tworzenie łańcuchów metod

get_included_graph_references

def get_included_graph_references() -> List[GraphDefinitionReference]

Pobierz listę dołączonych odwołań do grafów.

Zwraca:

• List[GraphDefinitionReference]: Lista odwołań do definicji grafu

describe

def describe(text: str) -> GraphSchema

Płynnie dodaj opis.

Parametry:

• text (str): Tekst opisu

Zwraca:

• GraphSchema: Samodzielne tworzenie łańcuchów metod

to_dict

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

Serializowanie schematu do słownika.

Zwraca:

• Dict[str, Any]: Schemat serializowany

to_json

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

Generowanie reprezentacji JSON.

Parametry:

• indent (int, default=2): poziom wcięcia JSON

Zwraca:

• str: ciąg JSON

to_gql

def to_gql() -> str

Generowanie definicji schematu GQL.

Zwraca:

• str: reprezentacja ciągu GQL

Metody klasy

create

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

Utwórz schemat grafu ze wszystkimi wymaganymi polami.

Parametry:

• name (str): nazwa schematu grafu
• nodes (List[Node], opcjonalnie): Definicje węzłów
• edges (List[Edge], opcjonalnie): Definicje krawędzi
• description (str, default=""): Opis schematu
• version (str, default="1.0"): wersja schematu

Zwraca:

• GraphSchema: Nowe wystąpienie schematu grafu

Klasy wejściowe zapytań

Klasy danych reprezentujące parametry wejściowe dla wstępnie zdefiniowanych zapytań grafów.

QueryInputBase

Klasa podstawowa dla wszystkich parametrów wejściowych zapytania.

Metody

to_json_payload

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

Konwertuj parametry wejściowe na słownik na potrzeby przesyłania interfejsu API.

Zwraca:

• Dict[str, Any]: Słownikowe reprezentacje parametrów wejściowych

validate

def validate() -> None

Zweryfikuj parametry wejściowe.

Podnosi:

• ValueError: Jeśli parametry wejściowe są nieprawidłowe

ReachabilityQueryInput

Parametry wejściowe zapytania o osiągalność między węzłami źródłowymi i docelowymi. Dziedziczy element, z ReachabilityQueryInputBase którego dziedziczy element .QueryInputBase

Pola

Sprawdzania poprawności:

• source_property_value jest wymagane
• target_property_value jest wymagane

Przykład:

# 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

Parametry wejściowe zapytania k-hop z danego węzła źródłowego. Dziedziczy po elem ReachabilityQueryInputBase.

Dziedziczy wszystkie pola z ReachabilityQueryInputprogramu .

Sprawdzania poprawności:

• Co najmniej jeden z source_property_value lub target_property_value musi zostać dostarczony

Przykład:

# 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

Parametry wejściowe zapytania promienia wybuchu od węzła źródłowego do docelowego. Dziedziczy po elem ReachabilityQueryInputBase.

Dziedziczy wszystkie pola z ReachabilityQueryInputprogramu z następującymi wymaganymi polami:

Sprawdzania poprawności:

• source_property_value jest wymagane
• target_property_value jest wymagane

Przykład:

# 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

Parametry wejściowe zapytania analizy centralności. Dziedziczy po elem QueryInputBase.

CentralityType, wyliczenie

Pola

Przykład:

# 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"]
))

RankingQueryInput

Parametry wejściowe zapytania analizy rankingowej. Dziedziczy po elem QueryInputBase.

Pola

Przykład:

# 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
))

Wyniki zapytania

QueryResult

Wynik zapytania grafowego z dostępem do leniwej ramki danych.

Konstruktor

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

Parametry:

• raw_response (Dict[str, Any]): Słownik odpowiedzi nieprzetworzonego interfejsu API
• graph (Graf): odwołanie do nadrzędnego grafu

Uwaga: Zazwyczaj tworzone przez Graph.query()program nie są tworzone bezpośrednio.

Metody

to_dataframe

def to_dataframe() -> DataFrame

Konwertuje wynik zapytania na ramkę danych platformy Spark.

Zwraca:

• DataFrame: Wynik zapytania jako ramka danych platformy Spark

Podnosi:

• ValueError: Jeśli konwersja zakończy się niepowodzeniem

Przykład:

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

get_raw_data

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

Pobierz sekcję RawData z odpowiedzi.

Zwraca:

• Dict[str, Any]: Słownik z nieprzetworzonymi metadanymi lub pustym dictem, jeśli nie istnieje

Przykład:

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

show

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

Wyświetl wynik zapytania w różnych formatach.

Parametry:

• format (str, default="visual"): format danych wyjściowych
  • "table": Pełne tabele ramek danych (wszystkie kolumny)
  • "visual": Interaktywna wizualizacja grafu z wtyczką VSC
  • "all": Pokaż wszystkie formaty

Podnosi:

• ValueError: Jeśli format nie jest jedną z obsługiwanych wartości

Przykład:

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

Kompletny przykład (zalecane — wersja 0.3 lub nowsza)

# 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()

Uwagi dotyczące wzorców projektowych

Płynny interfejs API

Wszyscy konstruktorzy obsługują tworzenie łańcuchów metod dla czytelnych, deklaratywnych definicji grafu:

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

Schematy unii

Wiele krawędzi o tym samym aliasie jest automatycznie połączonych ze scalonymi właściwościami:

# 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")

Autokonfiguracji

Wiele pól ma rozsądne wartości domyślne:

• Etykiety węzłów/krawędzi są domyślne dla ich aliasów
• Właściwości są automatycznie wnioskowane ze schematów źródłowych
• Domyślne grupy jednostek to etykiety główne/typy relacji

Ocena z opóźnieniem

Ramki danych i zasoby są ładowane w sposób leniwy i buforowane:

• graph_spec.nodes i graph_spec.edges są ładowane przy pierwszym dostępie
• Wyniki zapytania tworzą ramki danych tylko w przypadku żądania

Zawartość pokrewna

• Omówienie wykresów niestandardowych
• Tworzenie wykresów niestandardowych