Graph Builder-API-Referenz (Vorschau)

Die sentinel_graph-Klasse bietet eine Möglichkeit zur Interaktion mit dem Microsoft Sentinel Graphen, sodass Sie Ihr Graphschema definieren, Daten aus dem Microsoft Sentinel Data Lake in Knoten und Kanten transformieren, einen Graphen veröffentlichen und einen Abfragegraphen ausführen und erweiterte Graphalgorithmen ausführen können. Diese Klasse ist für die Verwendung der Spark-Sitzungen in Jupyter Notebooks konzipiert, die auf Microsoft Sentinel Spark-Compute ausgeführt werden.

GraphSpecBuilder

Die GraphSpecBuilder-Klasse bietet einen Fluent-Generator zum Erstellen von Graphspezifikationen mit Datenpipelines und Schemaintegration.

Wichtig

Der GraphBuilder Alias für diese Klasse ist veraltet und wird in einer zukünftigen Version entfernt. Verwenden Sie GraphSpecBuilder im gesamten neuen Code.

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

# Recommended
from sentinel_graph import GraphSpecBuilder

Constructor

GraphSpecBuilder(context: ExecutionContext)

Parameter:

context (ExecutionContext): Ausführungskontext mit Spark-Sitzung und -Konfiguration

Löst:

ValueError: Wenn der Kontext Keine ist oder der Graphname nicht bestimmt werden kann

Statische Methoden

`start`

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

Definieren Sie einen neuen Fluent-Graph-Generator.

Parameter:

context(ExecutionContext, optional): ExecutionContext instance. Bei "None" wird der Standardkontext verwendet.

Gibt:

GraphSpecBuilder: Neuer Generator instance

Beispiel:

builder = GraphSpecBuilder.start(context=context)

Instanzmethoden

`add_node`

def add_node(alias: str) -> NodeBuilderInitial

Beginnen Sie mit dem Erstellen einer Knotendefinition.

Parameter:

alias (str): Eindeutiger Bezeichner für diesen Knoten innerhalb des Graphen

Gibt:

NodeBuilderInitial: Knoten-Generator im Anfangszustand

Beispiel:

builder.add_node("user")

`add_edge`

def add_edge(alias: str) -> EdgeBuilderInitial

Beginnen Sie mit dem Erstellen einer Edgedefinition.

Parameter:

alias (str): Bezeichner für diesen Rand innerhalb des Graphen (kann über mehrere Kanten hinweg freigegeben werden)

Gibt:

EdgeBuilderInitial: Edge-Generator im Anfangszustand

Beispiel:

builder.add_edge("accessed")

`done`

def done() -> GraphSpec

Schließen Sie die Graphspezifikation ab, und geben Sie graphSpec instance zurück.

Gibt:

GraphSpec: Vollständige Graphspezifikation mit Datenpipeline und Schema

Löst:

ValueError: Wenn graph keine Knoten oder Kanten aufweist oder wenn die Überprüfung fehlschlägt

Beispiel:

graph_spec = builder.done()

GraphSpec

Graphspezifikation mit Datenpipeline-, Schema- und Anzeigefunktionen.

Constructor

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

Parameter:

name (str): Graphname
context (ExecutionContext): Ausführungskontext
graph_schema (GraphSchema): Graphschemadefinition
etl_pipeline (ETLPipeline, optional): Datenpipeline für die Graphvorbereitung

Eigenschaften

`nodes`

def nodes() -> DataFrame

Abrufen von Knotendatenrahmen (verzögert, zwischengespeichert). Bestimmt automatisch die Quelle aus der Datenpipeline oder Lake-Tabelle.

Gibt:

DataFrame: Spark DataFrame mit allen Knoten

Löst:

ValueError: Wenn Kontext fehlt oder DataFrames nicht geladen werden können

`edges`

def edges() -> DataFrame

Abrufen von Edgedatenrahmen (verzögert, zwischengespeichert). Bestimmt automatisch die Quelle aus der Datenpipeline oder Lake-Tabelle.

Gibt:

DataFrame: Spark DataFrame mit allen Kanten

Löst:

ValueError: Wenn Kontext fehlt oder DataFrames nicht geladen werden können

Methoden

`build_graph_with_data`

Hinweis

build_graph_with_data ist veraltet und wird in einer zukünftigen Version entfernt. Verwenden Sie Graph.build(spec) stattdessen .

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

Führen Sie die Datenpipeline aus, und veröffentlichen Sie den Graphen. Ruft intern auf Graph.build(self), stasht das zurückgegebene Graphund gibt ein abwärtskompatibles Wörterbuch zurück.

Gibt:

Dict[str, Any]: Wörterbuch mit:
- etl_result: Datenaufbereitungsergebnisse
- api_result: Ergebnisse veröffentlichen (falls erfolgreich)
- api_error: Fehlerzeichenfolge (wenn die Veröffentlichung fehlgeschlagen ist)
- instance_name: Graph instance Name
- status: "published" oder "prepared"

Beispiel:

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

`get_schema`

def get_schema() -> GraphSchema

Rufen Sie das Diagrammschema ab.

Gibt:

GraphSchema: Graphschemadefinition

`get_pipeline`

Hinweis

Diese Methode ist veraltet und wird in einer zukünftigen Version entfernt. Die Datenpipeline ist ein internes Implementierungsdetails, auf das nicht direkt zugegriffen werden sollte.

def get_pipeline() -> Optional[ETLPipeline]

Rufen Sie die Datenpipeline ab (keine für vorhandene Diagramme).

Gibt:

ETLPipeline oder None: Datenpipeline( falls verfügbar)

`to_graphframe`

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

Konvertieren des gesamten Graphen in GraphFrame zum Ausführen von Graphalgorithmen. Arbeitet nur mit lokalen Daten (aus Datenpipeline oder Lake-Tabelle).

Parameter:

column_mapping (Dict[str, str], optional): Benutzerdefinierte Spaltenzuordnung mit Schlüsseln:
- "id": Spaltenname der Scheitelpunkt-ID
- "source_id": Spaltenname der Edgequell-ID
- "target_id": Spaltenname der Edgeziel-ID

Gibt:

GraphFrame: GraphFrame-Objekt mit allen Scheitelpunkten und Kanten

Löst:

ValueError: Wenn ExecutionContext nicht verfügbar ist

Beispiel:

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

`show`

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

Anzeigen von Diagrammdaten in verschiedenen Formaten.

Parameter:

limit (int, default=100): Maximale Anzuzeigende Knoten/Kanten
viz_format (str, default="visual"): Ausgabeformat
- "table": Vollständige DataFrame-Tabellen (alle Spalten)
- "visual": Interaktive Graphvisualisierung
- "all": Alle Formate anzeigen

Löst:

ValueError: Wenn format nicht einer der unterstützten Werte ist

Beispiel:

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

`show_schema`

def show_schema() -> None

Zeigen Sie das Diagrammschema als interaktive Graphvisualisierung an.

Beispiel:

spec.show_schema()

Graph

Instance des abfragbaren Graphen. Erstellt über Graph.get() (vorhandenes Diagramm) oder Graph.build() (aus einem GraphSpec).

Constructor

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

Parameter:

name (str): Graphname
context (ExecutionContext): Ausführungskontext
spec (GraphSpec, optional): Angefügte Graphspezifikation (festgelegt durch Graph.build())
build_status (BuildStatus, optional): Buildergebnismetadaten (festgelegt durch Graph.build())

Löst:

ValueError: Wenn ExecutionContext den Wert None hat

Statische Methoden

`get`

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

Abrufen eines Diagramms instance aus einem vorhandenen Diagramm. Der zurückgegebene Graph verfügt über spec=None und build_status=None.

Parameter:

name(str): Graph instance Name
context (ExecutionContext, optional): Ausführungskontext (standardwert: ExecutionContext.default())

Gibt:

Graph: Graph instance

Löst:

ValueError: Wenn der Graphname leer ist oder der Graph instance nicht vorhanden ist

Beispiel:

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

`prepare`

Graph.prepare(spec: GraphSpec) -> Graph

Führen Sie die Datenaufbereitungsphase für eine ohne Veröffentlichung ausGraphSpec. Verwenden Sie publish() anschließend , um das Diagramm zu registrieren und abzufragen.

Parameter:

spec (GraphSpec): Graphspezifikation zur Vorbereitung

Gibt:

Graph: Graph instance mit spec angefügten undbuild_status.status == "prepared"

Löst:

ValueError: Wenn die Spezifikation keine Datenpipeline oder keinen Ausführungskontext aufweist
RuntimeError: Wenn die Ausführung der Datenpipeline fehlschlägt

Beispiel:

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

Erstellen Sie einen Graphen aus einem GraphSpec , indem Sie Daten vorbereiten und veröffentlichen. Ruft intern auf Graph.prepare(spec) und versucht graph.publish()dann . Im Gegensatz zum separaten Aufrufen dieser beiden Methoden werden Veröffentlichungsfehler abgefangen– das zurückgegebene Diagramm hat build_status.status == "prepared" und festgelegt, build_status.api_error anstatt ihn anzuheben.

Parameter:

spec (GraphSpec): Graphspezifikation, aus der erstellt werden soll

Gibt:

Graph: Graph instance mit spec angefügten und build_status aufgefüllten

Löst:

ValueError: Wenn die Spezifikation keine Datenpipeline oder keinen Ausführungskontext aufweist
RuntimeError: Wenn die Ausführung der Datenpipeline fehlschlägt

Beispiel:

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

Eigenschaften

`nodes`

def nodes() -> Optional[DataFrame]

Abrufen des DataFrame-Knotens. Delegiert an self.spec.nodes , wenn eine Spezifikation angefügt wird, andernfalls wird zurückgegeben None .

`edges`

def edges() -> Optional[DataFrame]

Abrufen von Edges-DataFrame. Delegiert an self.spec.edges , wenn eine Spezifikation angefügt wird, andernfalls wird zurückgegeben None .

`schema`

def schema() -> Optional[GraphSchema]

Abrufen des Diagrammschemas. Delegiert an self.spec.get_schema() , wenn eine Spezifikation angefügt wird, andernfalls wird zurückgegeben None .

Methoden

`query`

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

Führen Sie eine Abfrage für das Diagramm instance mithilfe von GQL aus.

Parameter:

query_string (str): Graphabfragezeichenfolge (GQL-Sprache)
query_language (str, default="GQL"): Abfragesprache

Gibt:

QueryResult: Objekt, das Knoten, Kanten und Metadaten enthält

Löst:

ValueError: Wenn ExecutionContext oder Spark-Sitzung fehlt
RuntimeError: Fehler bei der Clientinitialisierung oder Abfrageausführung

Beispiel:

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

[! HINWEIS] reachability(query_input=ReachabilityQueryInput(...)) wird weiterhin akzeptiert, gibt aber aus und wird in einer zukünftigen DeprecationWarning Version entfernt.

Führen Sie eine Analyse der Erreichbarkeit zwischen Quell- und Zielknoten durch.

Parameter:

source_property_value (str): Wert, der für die Quelleigenschaft übereinstimmen soll (zur Laufzeit überprüft. Muss bereitgestellt werden, wenn nicht query_inputverwendet wird.
target_property_value (str): Wert, der für die Zieleigenschaft übereinstimmen soll (zur Laufzeit überprüft. Muss bereitgestellt werden, wenn nicht query_inputverwendet wird.
source_property (Optional[str]): Eigenschaftenname zum Filtern von Quellknoten
participating_source_node_labels (Optional[List[str]]): Knotenbeschriftungen, die als Quellen berücksichtigt werden sollen
target_property (Optional[str]): Eigenschaftenname zum Filtern von Zielknoten
participating_target_node_labels (Optional[List[str]]): Knotenbeschriftungen, die als Ziele berücksichtigt werden sollen
participating_edge_labels (Optional[List[str]]): Zu durchlaufende Edgebeschriftungen
is_directional (bool): Gibt an, ob Kanten direktional sind (Standard: True)
min_hop_count (int): Minimale Hops (Standard: 1)
max_hop_count (int): Maximale Anzahl von Hops (Standard: 4)
shortest_path (bool): Nur kürzeste Pfade zurückgeben (Standard: False)
max_results (int): Maximale Ergebnisse (Standard: 500)

Löst:

ValueError: Wenn source_property_value oder target_property_value fehlt, min_hop_count < 1, max_hop_count < min_hop_countoder max_results < 1
RuntimeError: Fehler bei der Clientinitialisierung oder Abfrageausführung

Gibt:

QueryResult: Enthält die Erreichbarkeitspfade.

Beispiel:

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

Hinweis

k_hop(query_input=K_HopQueryInput(...)) wird weiterhin akzeptiert, gibt aber aus DeprecationWarning und wird in einer zukünftigen Version entfernt.

Führen Sie eine K-Hop-Analyse von einem bestimmten Quellknoten aus.

Parameter:

Identisch mit reachability

Validierung:

Mindestens eines von source_property_value oder target_property_value muss bereitgestellt werden.

Löst:

ValueError: Wenn weder source_property_value noch target_property_value bereitgestellt wird, oder wenn numerische Einschränkungen verletzt werden (identisch mit reachability)
RuntimeError: Fehler bei der Clientinitialisierung oder Abfrageausführung

Gibt:

QueryResult: Enthält die k-Hop-Ergebnisse

Beispiel:

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

Hinweis

blast_radius(query_input=BlastRadiusQueryInput(...)) wird weiterhin akzeptiert, gibt aber aus DeprecationWarning und wird in einer zukünftigen Version entfernt.

Führen Sie eine Analyse des Strahlradius vom Quellknoten zum Zielknoten durch.

Parameter:

source_property_value (str): Wert, der den Quellknoten identifiziert (zur Laufzeit überprüft). Muss bereitgestellt werden, wenn nicht query_inputverwendet wird.
target_property_value (str): Wert, der den Zielknoten identifiziert (zur Laufzeit überprüft). Muss bereitgestellt werden, wenn nicht query_inputverwendet wird.
Andere Parameter: identisch mit reachability

Löst:

ValueError: Wenn source_property_value oder target_property_value fehlt, oder wenn numerische Einschränkungen verletzt werden (identisch mit reachability)
RuntimeError: Fehler bei der Clientinitialisierung oder Abfrageausführung

Gibt:

QueryResult: Enthält die Ergebnisse des Strahlradius

Beispiel:

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

Hinweis

centrality(query_input=CentralityQueryInput(...)) wird weiterhin akzeptiert, gibt aber aus DeprecationWarning und wird in einer zukünftigen Version entfernt.

Führen Sie eine Zentralitätsanalyse für das Diagramm durch.

Parameter:

participating_source_node_labels (Optional[List[str]]): Quellknotenbeschriftungen
participating_target_node_labels (Optional[List[str]]): Zielknotenbeschriftungen
participating_edge_labels (Optional[List[str]]): Zu durchlaufende Edgebeschriftungen
threshold (int): Minimale Zentralitätsbewertung (Standard: 3); muss nicht negativ sein
centrality_type (CentralityType): CentralityType.Node oder CentralityType.Edge (Standard: None, fallback auf CentralityType.Node)
max_paths (int): Maximal zu berücksichtigende Pfade (Standard: 1000000; 0 = alle Pfade); muss nicht negativ sein.
is_directional (bool): Gibt an, ob Kanten direktional sind (Standard: True)
min_hop_count (int): Mindesthops (Standard: 1); muss ≥ 1 sein.
max_hop_count (int): Maximale Anzahl von Hops (Standard: 4); muss ≥ min_hop_count
shortest_path (bool): Nur kürzeste Pfade zurückgeben (Standard: False)
max_results (int): Maximale Ergebnisse (Standard: 500); muss ≥ 1 sein

Löst:

ValueError: Wenn threshold < 0, max_paths < 0, min_hop_count < 1, max_hop_count < min_hop_countoder max_results < 1
RuntimeError: Fehler bei der Clientinitialisierung oder Abfrageausführung

Gibt:

QueryResult: Enthält die Zentralitätsmetriken

Beispiel:

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

Hinweis

ranked(query_input=RankedQueryInput(...)) wird weiterhin akzeptiert, gibt aber aus DeprecationWarning und wird in einer zukünftigen Version entfernt.

Führen Sie eine Rangfolgeanalyse für das Diagramm aus.

Parameter:

rank_property_name (str): Für die Rangfolge zu verwendende Eigenschaftsname (zur Laufzeit überprüft). Muss bereitgestellt werden, wenn nicht query_inputverwendet wird.
threshold (int): Nur Rückgabepfade über dieser Gewichtung (Standard: 0); dürfen nicht negativ sein.
max_paths (int): Maximal zu berücksichtigende Pfade (Standard: 1000000; 0 = alle Pfade); muss nicht negativ sein.
decay_factor (float): Rangverfall pro Schritt; 2 bedeutet Halbierung (Standard: 1); muss nicht negativ sein.
is_directional (bool): Gibt an, ob Kanten direktional sind (Standard: True)
min_hop_count (int): Mindesthops (Standard: 1); muss ≥ 1 sein.
max_hop_count (int): Maximale Anzahl von Hops (Standard: 4); muss ≥ min_hop_count
shortest_path (bool): Nur kürzeste Pfade zurückgeben (Standard: False)
max_results (int): Maximale Ergebnisse (Standard: 500); muss ≥ 1 sein

Löst:

ValueError: Wenn rank_property_name fehlt, threshold < 0, max_paths < 0, decay_factor < 0, min_hop_count < 1, max_hop_count < min_hop_countoder max_results < 1
RuntimeError: Fehler bei der Clientinitialisierung oder Abfrageausführung

Gibt:

QueryResult: Enthält die bewerteten Knoten/Kanten.

Beispiel:

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

Konvertieren des gesamten Graphen in GraphFrame. Verwendet Spezifikationsdaten, sofern verfügbar; liest andernfalls aus Lake-Tabellen.

Parameter:

column_mapping (Dict[str, str], optional): Benutzerdefinierte Spaltenzuordnung

Gibt:

GraphFrame: GraphFrame-Objekt mit allen Scheitelpunkten und Kanten

Beispiel:

gf = graph.to_graphframe()

`show`

def show() -> None

Diagramminformationen anzeigen. Delegiert an spec.show() , um eine umfangreiche Anzeige zu erzielen, wenn eine Spezifikation angefügt ist. Andernfalls werden minimale Informationen ausgegeben.

`show_schema`

def show_schema() -> None

Diagrammschema anzeigen. Delegiert an spec.show_schema() , wenn eine Spezifikation angefügt wird. Gibt eine Meldung aus, die angibt, dass andernfalls kein Schema verfügbar ist.

`publish` (neu in v0.3.3)

def publish() -> Graph

Registrieren Sie den Graphen bei der API, sodass es abfragbar ist. Rufen Sie dies nach Graph.prepare() (oder auf einem beliebigen Auf, Graph dem eine Spezifikation angefügt ist) auf, um den Graphen instance zu veröffentlichen.

Gibt:

Graph: Selbst für Methodenketten

Löst:

ValueError: Wenn keine Spezifikation angefügt ist oder Kontext fehlt
RuntimeError: Wenn die Veröffentlichung fehlschlägt

Beispiel:

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

BuildStatus

Datenklasse, die Metadaten aus einem Graph.build() Vorgang enthält.

Felder

Feld	Typ	Beschreibung
`etl_result`	`Any`	Ergebnis aus der `prepare` Phase (Datenpipelineausführung)
`api_result`	`Optional[Dict]`	Ergebnis aus der Veröffentlichungsphase (`None` wenn die Veröffentlichung fehlgeschlagen ist)
`api_error`	`Optional[str]`	Fehlermeldung, wenn bei der Veröffentlichung ein Fehler aufgetreten ist (`None` wenn die Veröffentlichung erfolgreich war)
`instance_name`	`str`	Name des graphen instance
`status`	`Optional[BuildStatusKind]`	`None`, `"published"` oder `"prepared"`

Baupfade

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)

Beispiel:

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

Knoten-Generatoren

NodeBuilderInitial

Anfangszustand für Knoten-Generator: Nur Datenquellenmethoden verfügbar.

Constructor

NodeBuilderInitial(alias: str, graph_builder: GraphSpecBuilder)

Hinweis: In der Regel über GraphSpecBuilder.add_node()erstellt, nicht direkt instanziiert.

Methoden

Hinweis

Die Verwendung von Unterstrichen _ beim Benennen von Knoten, Kanten oder Eigenschaften in einem benutzerdefinierten Graphen wird nicht unterstützt. Ein Ungültiger Anforderungsfehler wird zurückgegeben, wenn Unterstriche verwendet werden.

`from_table`

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

Legen Sie die Tabelle als Datenquelle mit intelligenter Datenbankauflösung fest.

Parameter:

table_name (str): Name der Tabelle (erforderlich)
database (str, optional): Expliziter Datenbankname (hat Vorrang vor dem Kontextstandard)

Gibt:

NodeBuilderSourceSet: Generator für die weitere Konfiguration

Löst:

ValueError: Wenn die Tabelle nicht gefunden wurde oder mehrere in Konflikt stehende Tabellen gefunden wurden

Datenbankauflösungsreihenfolge:

Expliziter database Parameter (höchste Rangfolge)
ExecutionContext.default_database
Durchsuchen aller Datenbanken (mit Konflikterkennung)

Beispiel:

builder.add_node("user").from_table("SigninLogs", database="security_db")

`from_dataframe`

def from_dataframe(dataframe: DataFrame) -> NodeBuilderSourceSet

Legen Sie Spark DataFrame als Datenquelle fest.

Parameter:

dataframe (DataFrame): Spark DataFrame

Gibt:

NodeBuilderSourceSet: Generator für die weitere Konfiguration

Beispiel:

df = spark.read.table("users")
builder.add_node("user").from_dataframe(df)

NodeBuilderSourceSet

Knoten-Generator nach dem Festlegen der Datenquelle: Verfügbare Konfigurationsmethoden.

Constructor

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

Hinweis: Wird intern von NodeBuilderInitial-Quellmethoden erstellt.

Methoden

`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

Wenden Sie die Zeitbereichsfilterung auf die Datenquelle des Knotens an.

Parameter:

time_column (str): Spaltenname, der Zeitstempeldaten enthält (erforderlich)
start_time (str oder datetime, optional): Startdatum ('20.10.25', '2025-10-20' oder datetime-Objekt)
end_time (str oder datetime, optional): Enddatum (dieselben Formate wie start_time)
lookback_hours (float, optional): Stunden, die von jetzt an zurückschauen

Gibt:

NodeBuilderSourceSet: Selbst für Methodenketten

Löst:

ValueError: Wenn die Zeitspalte im Quellschema nicht gefunden wurde

Zeitbereichslogik:

Wenn start_time und end_time angegeben sind: Verwenden Sie sie direkt.
Wenn nur lookback_hours angegeben: end=now, start=now-lookback_hours
Wenn nichts angegeben ist: keine Zeitfilterung
Wenn Start/Ende UND lookback_hours: Start/Ende haben Vorrang

Beispiel:

# 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

Legen Sie die Knotenbezeichnung fest (standardmäßig alias, wenn nicht aufgerufen wird).

Parameter:

label (str): Knotenbezeichnung

Gibt:

NodeBuilderSourceSet: Selbst für Methodenketten

Löst:

ValueError: Wenn bezeichnung bereits festgelegt ist

Beispiel:

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

`with_columns`

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

Konfigurieren Sie Spalten mit der erforderlichen Schlüssel- und Anzeigebezeichnung.

Parameter:

*columns (str): Spaltennamen, die eingeschlossen werden sollen (mindestens ein Erforderlicher)
key (str): Spaltenname, der als Schlüssel markiert werden soll (erforderlich, muss in Spalten enthalten sein)
display (str): Spaltenname, der als Anzeigewert markiert werden soll (erforderlich, muss in Spalten vorhanden sein, kann mit schlüssel identisch sein)

Gibt:

NodeBuilderSourceSet: Selbst für Methodenketten

Löst:

ValueError: Wenn die Überprüfung fehlschlägt (doppelte Spalten, fehlender Schlüssel/Anzeige usw.)

Hinweise:

Eigenschaften werden automatisch aus Spaltentypen erstellt.
Die Zeitfilterspalte wird automatisch hinzugefügt, wenn sie angegeben ist.
Eigenschaftentypen werden automatisch aus dem Quellschema abgeleitet.
Weitere Informationen finden Sie unter Einschränkungen.

Beispiel:

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

Schließen Sie diesen Knoten ab, und beginnen Sie mit der Erstellung eines weiteren Knotens.

Parameter:

alias (str): Alias für den neuen Knoten

Gibt:

NodeBuilderInitial: Neuer Knoten-Generator

Beispiel:

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

Schließen Sie diesen Knoten ab, und beginnen Sie mit dem Erstellen eines Edges.

Parameter:

alias (str): Alias für den Edge

Gibt:

EdgeBuilderInitial: Neuer Edge-Generator

Beispiel:

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

`done`

def done() -> GraphSpec

Schließen Sie diesen Knoten ab, und schließen Sie die Graphspezifikation ab.

Gibt:

GraphSpec: Vollständige Graphspezifikation

Beispiel:

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

Edge Builders

EdgeBuilderInitial

Anfangszustand für Edge-Generator: Nur Datenquellenmethoden verfügbar.

Constructor

EdgeBuilderInitial(alias: str, graph_builder: GraphSpecBuilder)

Hinweis: In der Regel über GraphSpecBuilder.add_edge()erstellt, nicht direkt instanziiert.

Methoden

Hinweis

`from_table`

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

Legen Sie die Tabelle als Datenquelle mit intelligenter Datenbankauflösung fest.

Parameter:

table_name (str): Name der Tabelle (erforderlich)
database (str, optional): Expliziter Datenbankname

Gibt:

EdgeBuilderSourceSet: Generator für die weitere Konfiguration

Löst:

ValueError: Wenn die Tabelle nicht gefunden wurde oder mehrere in Konflikt stehende Tabellen gefunden wurden

Beispiel:

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

`from_dataframe`

def from_dataframe(dataframe: DataFrame) -> EdgeBuilderSourceSet

Legen Sie Spark DataFrame als Datenquelle fest.

Parameter:

dataframe (DataFrame): Spark DataFrame

Gibt:

EdgeBuilderSourceSet: Generator für die weitere Konfiguration

Beispiel:

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

EdgeBuilderSourceSet

Edge-Generator nach dem Festlegen der Datenquelle: Verfügbare Konfigurationsmethoden.

Constructor

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

Hinweis: Wird intern von EdgeBuilderInitial-Quellmethoden erstellt.

Methoden

`with_label`

def with_label(label: str) -> EdgeBuilderSourceSet

Legen Sie den Typ/die Bezeichnung der Edgebeziehung fest (wird standardmäßig alias verwendet, wenn nicht aufgerufen wird).

Parameter:

label (str): Edge-Bezeichnung

Gibt:

EdgeBuilderSourceSet: Selbst für Methodenketten

Löst:

ValueError: Wenn bezeichnung bereits festgelegt ist

Beispiel:

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

`edge_label`

Hinweis

Verwenden Sie with_label() stattdessen . Diese Methode wird in einer zukünftigen Version entfernt.

def edge_label(label: str) -> EdgeBuilderSourceSet

Legen Sie den Typ/die Bezeichnung der Edgebeziehung fest (wird standardmäßig alias verwendet, wenn nicht aufgerufen wird).

Parameter:

label (str): Edge-Bezeichnung

Gibt:

EdgeBuilderSourceSet: Selbst für Methodenketten

Löst:

ValueError: Wenn bezeichnung bereits festgelegt ist

Beispiel:

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

`source`

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

Legen Sie den Quellknoten mit ID-Spalte und -Bezeichnung fest.

Parameter:

id_column (str): Spaltenname mit Quellknoten-ID
node_type (str): Bezeichnung des Quellknotens

Gibt:

EdgeBuilderSourceSet: Selbst für Methodenketten

Löst:

ValueError: Wenn quelle bereits festgelegt ist

Beispiel:

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

Legen Sie den Zielknoten mit ID-Spalte und Bezeichnung fest.

Parameter:

id_column (str): Spaltenname, der die Zielknoten-ID enthält
node_type (str): Zielknotenbezeichnung

Gibt:

EdgeBuilderSourceSet: Selbst für Methodenketten

Löst:

ValueError: Wenn das Ziel bereits festgelegt ist

Beispiel:

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

Wenden Sie die Zeitbereichsfilterung auf die Datenquelle des Edges an.

Parameter:

time_column (str): Spaltenname, der Zeitstempeldaten enthält (erforderlich)
start_time (str oder datetime, optional): Startdatum
end_time (str oder datetime, optional): Enddatum
lookback_hours (float, optional): Stunden, die von jetzt an zurückschauen

Gibt:

EdgeBuilderSourceSet: Selbst für Methodenketten

Löst:

ValueError: Wenn die Zeitspalte im Quellschema nicht gefunden wurde

Beispiel:

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

Konfigurieren Sie Spalten mit der erforderlichen Schlüssel- und Anzeigebezeichnung.

Parameter:

*columns (str): Spaltennamen, die eingeschlossen werden sollen (mindestens ein Erforderlicher)
key (str): Spaltenname, der als Schlüssel markiert werden soll (erforderlich, muss in Spalten enthalten sein)
display (str): Spaltenname, der als Anzeigewert markiert werden soll (erforderlich, muss in Spalten enthalten sein)

Gibt:

EdgeBuilderSourceSet: Selbst für Methodenketten

Löst:

ValueError: Wenn die Überprüfung fehlschlägt

Beispiel:

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

Schließen Sie diesen Edge ab, und beginnen Sie mit dem Erstellen eines Knotens.

Parameter:

alias (str): Alias für den neuen Knoten

Gibt:

NodeBuilderInitial: Neuer Knoten-Generator

`add_edge`

def add_edge(alias: str) -> EdgeBuilderInitial

Schließen Sie diesen Rand ab, und beginnen Sie mit dem Erstellen eines weiteren Edges.

Parameter:

alias (str): Alias für den neuen Edge

Gibt:

EdgeBuilderInitial: Neuer Edge-Generator

Beispiel:

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

Schließen Sie diesen Rand ab, und schließen Sie die Graphspezifikation ab.

Gibt:

GraphSpec: Vollständige Graphspezifikation

Schemaklassen

GraphDefinitionReference

Verweis auf eine Graphdefinition mit Name und Version.

Constructor

GraphDefinitionReference(
    fully_qualified_name: str,
    version: str
)

Parameter:

fully_qualified_name (str): Vollqualifizierter Name des Graphen, auf den verwiesen wird
version (str): Version des Graphen, auf das verwiesen wird

Löst:

ValueError: Wenn fully_qualified_name oder Version leer ist

Methoden

`to_dict`

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

Serialisieren sie in ein Wörterbuch.

Gibt:

Dict[str, Any]: Serialisierter Verweis

Eigenschaft

Eigenschaftendefinition mit typsicherer Schnittstelle.

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
)

Parameter:

name (str): Eigenschaftsname
property_type (PropertyType): Eigenschaftsdatentyp
is_non_null (bool, default=False): Gibt an, ob die Eigenschaft erforderlich ist.
description (str, default=""): Eigenschaftenbeschreibung
is_key (bool, default=False): Gibt an, ob die Eigenschaft ein Schlüssel ist.
is_display_value (bool, default=False): Gibt an, ob die Eigenschaft ein Anzeigewert ist.
is_internal (bool, default=False): Gibt an, ob die Eigenschaft intern ist.

Löst:

ValueError: Wenn name leer ist oder die Überprüfung fehlschlägt

Klassenmethoden

`key`

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

Erstellen Sie eine Schlüsseleigenschaft mit allgemeinen Einstellungen (is_key=True, is_display_value=True).

`display`

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

Erstellen Sie eine Anzeigewerteigenschaft (is_display_value=True).

Methoden

`describe`

def describe(text: str) -> Property

Fügen Sie eine Beschreibung fließend hinzu.

Parameter:

text (str): Beschreibungstext

Gibt:

Property: Selbst für Methodenketten

`to_dict`

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

Serialisieren Sie die Eigenschaft in ein Wörterbuch mit Den Anmerkungsschlüsseln mit dem Präfix @.

Gibt:

Dict[str, Any]: Serialisierte Eigenschaft

`to_gql`

def to_gql() -> str

Generieren sie die Definition der GQL-Eigenschaft.

Gibt:

str: GQL-Zeichenfolgendarstellung

EdgeNode

Knotenverweis, der in Edgedefinitionen verwendet wird.

Constructor

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

Parameter:

alias (str, optional): Knotenalias (automatisch auf erste Bezeichnung festgelegt, wenn Keine oder leer)
labels (List[str]): Knotenbezeichnungen (mindestens eine erforderlich)

Löst:

ValueError: Wenn die Bezeichnungsliste leer ist
TypeError: Wenn Bezeichnungen keine Zeichenfolgen sind

Automatische Mutation:

Wenn der Alias Keine oder leer ist, wird er auf die erste Bezeichnung festgelegt.

Methoden

`to_dict`

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

Serialisieren sie in ein Wörterbuch.

Gibt:

Dict[str, Any]: Serialisierter Edgeknotenverweis

Knoten

Knotendefinition mit typsicherer Schnittstelle.

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
)

Parameter:

alias (str, default=""):Knotenalias (automatisch auf erste Bezeichnung festgelegt, wenn leer)
labels (List[str]): Knotenbezeichnungen (mindestens eine erforderlich)
implies_labels (List[str], default=[]): Implizite Bezeichnungen
properties (List[Property], default=[]): Knoteneigenschaften
description (str, default=""): Knotenbeschreibung
entity_group (str, default=""): Name der Entitätsgruppe
dynamic_labels (bool, default=False): Gibt an, ob der Knoten über dynamische Bezeichnungen verfügt.
abstract_edge_aliases (bool, default=False): Gibt an, ob der Knoten abstrakte Edgealiase verwendet.

Löst:

ValueError: Wenn die Überprüfung fehlschlägt (keine Bezeichnungen, keine Schlüsseleigenschaft, keine Anzeigeeigenschaft usw.)

Automatische Mutation:

Wenn der Alias leer ist, wird er auf die erste Bezeichnung festgelegt.
Wenn entity_group leer ist, wird sie auf die primäre Bezeichnung festgelegt.

Methoden

`get_primary_label`

def get_primary_label() -> Optional[str]

Rufen Sie die primäre (erste) Bezeichnung ab.

Gibt:

str oder None: Primäre Bezeichnung oder Keine, wenn keine Bezeichnungen

`get_entity_group_name`

def get_entity_group_name() -> str

Rufen Sie den Namen der Entitätsgruppe oder das Fallback auf die primäre Bezeichnung ab.

Gibt:

str: Name der Entitätsgruppe

`get_primary_key_property_name`

def get_primary_key_property_name() -> Optional[str]

Rufen Sie den Namen der Primärschlüsseleigenschaft ab.

Gibt:

str oder None: Primärschlüsseleigenschaftsname

`get_properties`

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

Abrufen von Eigenschaften als Wörterbuch für einfachen Zugriff.

Gibt:

Dict[str, Property]: Eigenschaften mit Schlüsseln nach Namen

`get_property`

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

Rufen Sie eine bestimmte Eigenschaft anhand des Namens ab.

Parameter:

name (str): Eigenschaftsname

Gibt:

Property oder None: Eigenschaft, falls gefunden

`add_property`

def add_property(prop: Property) -> None

Fügen Sie diesem Knoten eine Eigenschaft hinzu.

Parameter:

prop (Eigenschaft): Hinzuzufügende Eigenschaft

Löst:

ValueError: Wenn der Eigenschaftsname dupliziert ist

`is_dynamically_labeled`

def is_dynamically_labeled() -> bool

Überprüfen Sie, ob der Knoten über dynamische Bezeichnungen verfügt.

Gibt:

bool: True, wenn dynamische Bezeichnungen aktiviert sind

`is_abstract_edge_node_aliases`

def is_abstract_edge_node_aliases() -> bool

Überprüfen Sie, ob der Knoten abstrakte Edgeknotenaliase verwendet.

Gibt:

bool: True, wenn abstrakte Edgealiase aktiviert sind

`describe`

def describe(text: str) -> Node

Fügen Sie eine Beschreibung fließend hinzu.

Parameter:

text (str): Beschreibungstext

Gibt:

Node: Selbst für Methodenketten

`to_dict`

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

Serialisieren des Knotens in ein Wörterbuch.

Gibt:

Dict[str, Any]: Serialisierter Knoten

`to_gql`

def to_gql() -> str

Generieren einer GQL-Knotendefinition.

Gibt:

str: GQL-Zeichenfolgendarstellung

Löst:

ValueError: Wenn der Knoten keine erforderlichen Felder für GQL enthält

Klassenmethoden

`create`

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

Erstellen Sie einen Knoten mit allen erforderlichen Feldern.

Parameter:

alias (str): Knotenalias
labels (List[str]): Knotenbeschriftungen
properties (List[Property]): Knoteneigenschaften
description (str, default=""): Knotenbeschreibung
entity_group (str, default=""): Name der Entitätsgruppe

Gibt:

Node: Neuer Knoten instance

Microsoft Edge

Edgedefinition mit typsicherer Schnittstelle.

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
)

Parameter:

relationship_type (str): Edgebeziehungstyp (z. B. "FOLLOWS", "OWNS")
source_node_label (str): Bezeichnung des Quellknotens
target_node_label (str): Zielknotenbezeichnung
direction (EdgeDirection, default=DIRECTED_RIGHT): Edge direction
properties (List[Property], default=[]): Edgeeigenschaften
description (str, default=""): Edge description
entity_group (str, default=""): Name der Entitätsgruppe
dynamic_type (bool, default=False): Gibt an, ob edge einen dynamischen Typ aufweist.

Löst:

ValueError: Wenn die Überprüfung fehlschlägt

Automatische Mutation:

labels list wird automatisch mit aufgefüllt. [relationship_type]
Wenn entity_group leer ist, wird sie auf relationship_type

Eigenschaften

`edge_type`

def edge_type() -> str

Abwärtskompatibilitätsalias für relationship_type.

Gibt:

str: Beziehungstyp

Methoden

`get_entity_group_name`

def get_entity_group_name() -> str

Rufen Sie den Namen der Entitätsgruppe oder das Fallback auf den Beziehungstyp ab.

Gibt:

str: Name der Entitätsgruppe

`is_dynamic_type`

def is_dynamic_type() -> bool

Überprüfen Sie, ob edge den dynamischen Typ aufweist.

Gibt:

bool: True, wenn dynamischer Typ

`add_property`

def add_property(edge_property: Property) -> None

Fügen Sie dieser Kante eine -Eigenschaft hinzu.

Parameter:

edge_property (Eigenschaft): Hinzuzufügende Eigenschaft

`describe`

def describe(text: str) -> Edge

Fügen Sie eine Beschreibung fließend hinzu.

Parameter:

text (str): Beschreibungstext

Gibt:

Edge: Selbst für Methodenketten

`to_dict`

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

Serialisieren Sie Edge in Wörterbuch.

Gibt:

Dict[str, Any]: Serialisierter Edge

`to_gql`

def to_gql() -> str

Generieren sie die GQL-Edgedefinition.

Gibt:

str: GQL-Zeichenfolgendarstellung

Klassenmethoden

`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

Erstellen Sie eine Kante mit allen erforderlichen Feldern.

Parameter:

relationship_type (str): Edge-Beziehungstyp
source_node_label (str): Bezeichnung des Quellknotens
target_node_label (str): Zielknotenbezeichnung
properties (List[Property], optional): Edge-Eigenschaften
description (str, default=""): Edge description
entity_group (str, default=""): Name der Entitätsgruppe

Gibt:

Edge: Neue Edge-instance

GraphSchema

Graphschemadefinition mit typsicherer Schnittstelle.

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

Parameter:

name (str): Graphschemaname
nodes (List[Node], default=[]): Knotendefinitionen
edges (List[Edge], default=[]): Edgedefinitionen
base_graphs (List[GraphSchema], default=[]): Basisdiagrammschemas
description (str, default=""): Schemabeschreibung
version (str, default="1.0"): Schemaversion
fully_qualified_name (str, default=""): Vollqualifizierter Name
namespace (str, default=""): Namespace

Löst:

ValueError: Wenn die Überprüfung fehlschlägt (doppelte Aliase, Edges verweisen auf nicht vorhandene Knoten usw.)

Methoden

`get_fully_qualified_name`

def get_fully_qualified_name() -> str

Rufen Sie den vollqualifizierten Namen ab.

Gibt:

str: Vollqualifizierter Name

`get_namespace`

def get_namespace() -> str

Rufen Sie den Namespace aus dem vollqualifizierten Namen ab oder geben Sie den Standardwert zurück.

Gibt:

str:Namespace

`get_version`

def get_version() -> str

Version abrufen.

Gibt:

str: Versionszeichenfolge

`get_node`

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

Ruft den Knoten nach Bezeichnung oder Alias ab.

Parameter:

label_or_alias (str): Knotenbezeichnung oder Alias

Gibt:

Node oder None: Knoten, falls gefunden

`get_edge`

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

Rufen Sie den Edge nach Name/Typ ab.

Parameter:

name (str): Edge-Beziehungstyp

Gibt:

Edge oder None: Edge, falls gefunden

`add_node`

def add_node(node: Node) -> None

Fügen Sie diesem Diagramm einen Knoten hinzu.

Parameter:

node (Knoten): Hinzuzufügende Knoten

Löst:

ValueError: Wenn der Knotenalias dupliziert ist

`add_edge`

def add_edge(edge: Edge) -> None

Fügen Sie diesem Diagramm eine Kante hinzu.

Parameter:

edge (Edge): Hinzuzufügende Edge

Löst:

ValueError: Wenn der Edgetyp dupliziert ist

`include_graph`

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

Hinzufügen eines Graph-Includes (Fluent-API).

Parameter:

fully_qualified_name (str): Vollqualifizierter Name des einzuschließenden Graphen
version (str): Version des einzuschließenden Graphen

Gibt:

GraphSchema: Selbst für Methodenketten

`get_included_graph_references`

def get_included_graph_references() -> List[GraphDefinitionReference]

Ruft eine Liste der enthaltenen Graphverweise ab.

Gibt:

List[GraphDefinitionReference]: Liste der Graphdefinitionsverweise

`describe`

def describe(text: str) -> GraphSchema

Fügen Sie eine Beschreibung fließend hinzu.

Parameter:

text (str): Beschreibungstext

Gibt:

GraphSchema: Selbst für Methodenketten

`to_dict`

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

Serialisieren Sie das Schema in ein Wörterbuch.

Gibt:

Dict[str, Any]: Serialisiertes Schema

`to_json`

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

Generieren einer JSON-Darstellung.

Parameter:

indent (int, default=2): JSON-Einzugsebene

Gibt:

str: JSON-Zeichenfolge

`to_gql`

def to_gql() -> str

Generieren einer GQL-Schemadefinition.

Gibt:

str: GQL-Zeichenfolgendarstellung

Klassenmethoden

`create`

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

Erstellen Sie ein Diagrammschema mit allen erforderlichen Feldern.

Parameter:

name (str): Graphschemaname
nodes (List[Node], optional): Knotendefinitionen
edges (List[Edge], optional): Edgedefinitionen
description (str, default=""): Schemabeschreibung
version (str, default="1.0"): Schemaversion

Gibt:

GraphSchema: Neue instance des Diagrammschemas

Abfrageeingabeklassen

Datenklassen, die Eingabeparameter für vordefinierte Graphabfragen darstellen.

Hinweis

Das direkte Übergeben von QueryInput Objekten an Graph Abfragemethoden ist veraltet und wird in zukünftigen Versionen entfernt. Verwenden Sie stattdessen Schlüsselwort (keyword) Argumente. Die Graph Methoden (reachability, , blast_radiusk_hop, centrality, ) rankedakzeptieren alle Parameter als Schlüsselwort (keyword) Argumente und erstellen die Eingabeobjekte intern. Diese Klassen verbleiben vorerst in der Codebasis, sollten aber nicht in neuem Code verwendet werden.

QueryInputBase

Basisklasse für alle Abfrageeingabeparameter.

Methoden

`to_json_payload`

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

Konvertieren Sie die Eingabeparameter für die API-Übermittlung in ein Wörterbuch.

Gibt:

Dict[str, Any]: Wörterbuchdarstellung der Eingabeparameter

`validate`

def validate() -> None

Überprüfen Sie die Eingabeparameter.

Löst:

ValueError: Wenn die Eingabeparameter ungültig sind

ReachabilityQueryInput

Eingabeparameter für eine Erreichbarkeitsabfrage zwischen Quell- und Zielknoten. Erbt von , von ReachabilityQueryInputBase dem von QueryInputBaseerbt.

Felder

Feld	Typ	Standard	Beschreibung
`source_property_value`	`str`	(erforderlich)	Wert, der für die Quelleigenschaft übereinstimmen soll
`target_property_value`	`str`	(erforderlich)	Wert, der für die Zieleigenschaft übereinstimmen soll
`source_property`	`Optional[str]`	`None`	Eigenschaftenname zum Filtern von Quellknoten
`participating_source_node_labels`	`Optional[List[str]]`	`None`	Knotenbezeichnungen, die als Quellknoten berücksichtigt werden sollen
`target_property`	`Optional[str]`	`None`	Eigenschaftenname zum Filtern von Zielknoten
`participating_target_node_labels`	`Optional[List[str]]`	`None`	Knotenbezeichnungen, die als Zielknoten berücksichtigt werden sollen
`participating_edge_labels`	`Optional[List[str]]`	`None`	Edgebeschriftungen, die im Pfad durchlaufen werden sollen
`is_directional`	`Optional[bool]`	`True`	Gibt an, ob die Kanten direktional sind
`min_hop_count`	`Optional[int]`	`1`	Minimale Anzahl von Hops im Pfad
`max_hop_count`	`Optional[int]`	`4`	Maximale Anzahl von Hops im Pfad
`shortest_path`	`Optional[bool]`	`False`	Ob nur der kürzeste Pfad gefunden werden soll
`max_results`	`Optional[int]`	`500`	Maximale Anzahl von Ergebnissen, die zurückgegeben werden sollen

Validierung:

source_property_value ist erforderlich.
target_property_value ist erforderlich.

Beispiel:

# 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

Eingabeparameter für eine K-Hop-Abfrage von einem bestimmten Quellknoten. Erbt von ReachabilityQueryInputBase.

Erbt alle Felder von ReachabilityQueryInput.

Validierung:

Mindestens eines von source_property_value oder target_property_value muss bereitgestellt werden.

Beispiel:

# 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

Eingabeparameter für eine Abfrage des Strahlradius von der Quelle zu den Zielknoten. Erbt von ReachabilityQueryInputBase.

Erbt alle Felder von ReachabilityQueryInputmit den folgenden Pflichtfeldern:

Feld	Typ	Erforderlich	Beschreibung
`source_property_value`	`str`	Ja	Wert zum Identifizieren des Quellknotens
`target_property_value`	`str`	Ja	Wert zum Identifizieren des Zielknotens

Validierung:

source_property_value ist erforderlich.
target_property_value ist erforderlich.

Beispiel:

# 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

Eingabeparameter für eine Zentralitätsanalyseabfrage. Erbt von QueryInputBase.

CentralityType-Enumeration

Wert	Beschreibung
`CentralityType.Node`	Zentralität von Computeknoten
`CentralityType.Edge`	Compute-Edge-Zentralität

Felder

Feld	Typ	Standard	Beschreibung
`threshold`	`Optional[int]`	`3`	Zu berücksichtigende Minimale Zentralitätsbewertung
`centrality_type`	`CentralityType`	`CentralityType.Node`	Typ der zentral zu berechnenden Zentralität
`max_paths`	`Optional[int]`	`1000000`	Maximal zu berücksichtigende Pfade (0 = alle)
`participating_source_node_labels`	`Optional[List[str]]`	`None`	Quellknotenbezeichnungen
`participating_target_node_labels`	`Optional[List[str]]`	`None`	Zielknotenbezeichnungen
`participating_edge_labels`	`Optional[List[str]]`	`None`	Zu durchlaufende Kantenbeschriftungen
`is_directional`	`Optional[bool]`	`True`	Gibt an, ob Ränder direktional sind
`min_hop_count`	`Optional[int]`	`1`	Mindestanzahl von Hops
`max_hop_count`	`Optional[int]`	`4`	Maximale Anzahl von Hops
`shortest_path`	`Optional[bool]`	`False`	Nur kürzeste Pfade
`max_results`	`Optional[int]`	`500`	Maximale Ergebnisse

Beispiel:

# 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

Eingabeparameter für eine Rangfolgeanalyseabfrage. Erbt von QueryInputBase.

Felder

Feld	Typ	Standard	Beschreibung
`rank_property_name`	`str`	(erforderlich)	Eigenschaftsname, der für Rangfolgepfade verwendet werden soll
`threshold`	`Optional[int]`	`0`	Nur Pfade mit Gewichtungen über diesem Wert zurückgeben
`max_paths`	`Optional[int]`	`1000000`	Maximal zu berücksichtigende Pfade (0 = alle)
`decay_factor`	`Optional[float]`	`1`	Wie viel die einzelnen Diagrammschritte den Rang reduzieren (2 = halbiert jeden Schritt)
`is_directional`	`Optional[bool]`	`True`	Gibt an, ob Ränder direktional sind
`min_hop_count`	`Optional[int]`	`1`	Mindestanzahl von Hops
`max_hop_count`	`Optional[int]`	`4`	Maximale Anzahl von Hops
`shortest_path`	`Optional[bool]`	`False`	Nur kürzeste Pfade
`max_results`	`Optional[int]`	`500`	Maximale Ergebnisse

Beispiel:

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

Query Results

QueryResult

Ergebnis einer Graphabfrage mit verzögertem DataFrame-Zugriff.

Constructor

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

Parameter:

raw_response (Dict[str, Any]): Unformatiertes API-Antwortwörterbuch
graph (Graph): Verweis auf übergeordnetes Diagramm

Hinweis: In der Regel von Graph.query()erstellt, nicht direkt instanziiert.

Methoden

`to_dataframe`

def to_dataframe() -> DataFrame

Konvertiert das Abfrageergebnis in einen Spark-DataFrame.

Gibt:

DataFrame: Abfrageergebnis als Spark DataFrame

Löst:

ValueError: Wenn die Konvertierung fehlschlägt

Beispiel:

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

`get_raw_data`

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

Rufen Sie den RawData-Abschnitt aus der Antwort ab.

Gibt:

Dict[str, Any]: Wörterbuch mit unformatierten Metadaten oder leeres Diktat, falls nicht vorhanden

Beispiel:

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

`show`

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

Anzeigen des Abfrageergebnisses in verschiedenen Formaten.

Parameter:

format (str, default="visual"): Ausgabeformat
- "table": Vollständige DataFrame-Tabellen (alle Spalten)
- "visual": Interaktive Graphvisualisierung mit VSC-Plug-In
- "all": Alle Formate anzeigen

Löst:

ValueError: Wenn format nicht einer der unterstützten Werte ist

Beispiel:

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

Vollständiges Beispiel (empfohlen – v0.3 und höher)

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

Hinweise zu Entwurfsmustern

Fluent-API

Alle Generatoren unterstützen die Methodenkette für lesbare, deklarative Graphdefinitionen:

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

Union-Schemas

Mehrere Kanten mit demselben Alias werden automatisch mit zusammengeführten Eigenschaften verbunden:

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

Autokonfiguration

Viele Felder weisen sinnvolle Standardwerte auf:

Knoten-/Edgebezeichnungen werden standardmäßig auf ihre Aliase festgelegt.
Eigenschaften werden automatisch aus Quellschemas abgeleitet
Entitätsgruppen verwenden standardmäßig primäre Bezeichnungen/Beziehungstypen

Verzögerte Auswertung

DataFrames und Ressourcen werden verzögert geladen und zwischengespeichert:

graph_spec.nodes und graph_spec.edges werden beim ersten Zugriff geladen.
Abfrageergebnisse erstellen DataFrames nur bei Bedarf

Feedback

War diese Seite hilfreich?

Last updated on 2026-04-23

Freigeben über

Graph Builder-API-Referenz (Vorschau)

GraphSpecBuilder

Constructor

Statische Methoden

start

Instanzmethoden

add_node

add_edge

done

GraphSpec

Constructor

Eigenschaften

nodes

edges

Methoden

build_graph_with_data

get_schema

get_pipeline

to_graphframe

show

show_schema

Graph

Constructor

Statische Methoden

get

prepare

build

Eigenschaften

nodes

edges

schema

Methoden

query

reachability

k_hop

blast_radius

centrality

ranked

to_graphframe

show

show_schema

publish (neu in v0.3.3)

BuildStatus

Felder

Baupfade

Knoten-Generatoren

NodeBuilderInitial

Constructor

Methoden

from_table

from_dataframe

NodeBuilderSourceSet

Constructor

Methoden

with_time_range

with_label

with_columns

add_node

add_edge

done

Edge Builders

EdgeBuilderInitial

Constructor

Methoden

from_table

from_dataframe

EdgeBuilderSourceSet

Constructor

Methoden

with_label

edge_label

source

target

with_time_range

with_columns

add_node

add_edge

done

Schemaklassen

`start`

`add_node`

`add_edge`

`done`

`nodes`

`edges`

`build_graph_with_data`

`get_schema`

`get_pipeline`

`to_graphframe`

`show`

`show_schema`

`get`

`prepare`

`build`

`nodes`

`edges`

`schema`

`query`

`reachability`

`k_hop`

`blast_radius`

`centrality`

`ranked`

`to_graphframe`

`show`

`show_schema`

`publish` (neu in v0.3.3)

`from_table`

`from_dataframe`

`with_time_range`

`with_label`

`with_columns`

`add_node`

`add_edge`

`done`

`from_table`

`from_dataframe`

`with_label`

`edge_label`

`source`

`target`

`with_time_range`

`with_columns`

`add_node`

`add_edge`

`done`

`to_dict`

`key`

`display`

`describe`

`to_dict`

`to_gql`

`to_dict`

`get_primary_label`

`get_entity_group_name`

`get_primary_key_property_name`

`get_properties`

`get_property`

`add_property`

`is_dynamically_labeled`

`is_abstract_edge_node_aliases`

`describe`

`to_dict`

`to_gql`

`create`

`edge_type`

`get_entity_group_name`

`is_dynamic_type`

`add_property`

`describe`

`to_dict`

`to_gql`

`create`

`get_fully_qualified_name`

`get_namespace`

`get_version`

`get_node`

`get_edge`

`add_node`