Graph Builder API-verwijzing (preview)

De klasse sentinel_graph biedt een manier om te communiceren met de Microsoft Sentinel-grafiek, zodat u uw grafiekschema kunt definiëren, gegevens van de Microsoft Sentinel Data Lake kunt transformeren in knooppunten en randen, een grafiek kunt publiceren, een query kunt uitvoeren op grafiek en geavanceerde grafiekalgoritmen kunt uitvoeren. Deze klasse is ontworpen voor gebruik met de Spark-sessies in Jupyter-notebooks die worden uitgevoerd op Microsoft Sentinel Spark-berekening.

GraphSpecBuilder

De GraphSpecBuilder-klasse biedt een vloeiende opbouwfunctie voor het maken van grafiekspecificaties met gegevenspijplijnen en schema-integratie.

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

# Recommended
from sentinel_graph import GraphSpecBuilder

Constructor

GraphSpecBuilder(context: ExecutionContext)

Parameters:

• context (ExecutionContext): Uitvoeringscontext met Spark-sessie en -configuratie

Verhoogt:

• ValueError: Als de context Geen of grafieknaam is, kan niet worden bepaald

Statische methoden

start

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

Definieer een nieuwe opbouwfunctie voor Fluent Graph.

Parameters:

• context (ExecutionContext, optioneel): ExecutionContext-exemplaar. Als geen, gebruikt standaardcontext.

Retourneert:

• GraphSpecBuilder: Nieuw builder-exemplaar

Voorbeeld:

builder = GraphSpecBuilder.start(context=context)

Exemplaarmethoden

add_node

def add_node(alias: str) -> NodeBuilderInitial

Begin met het bouwen van een knooppuntdefinitie.

Parameters:

• alias (str): unieke id voor dit knooppunt in de grafiek

Retourneert:

• NodeBuilderInitial: Opbouwfunctie voor knooppunten in initiële status

Voorbeeld:

builder.add_node("user")

add_edge

def add_edge(alias: str) -> EdgeBuilderInitial

Begin met het bouwen van een edge-definitie.

Parameters:

• alias (str): id voor deze rand in de grafiek (kan worden gedeeld over meerdere randen)

Retourneert:

• EdgeBuilderInitial: Edge Builder in initiële status

Voorbeeld:

builder.add_edge("accessed")

done

def done() -> GraphSpec

Grafiekspecificatie voltooien en GraphSpec-exemplaar retourneren.

Retourneert:

• GraphSpec: Volledige grafiekspecificatie met gegevenspijplijn en schema

Verhoogt:

• ValueError: Als de grafiek geen knooppunten of randen heeft of als de validatie mislukt

Voorbeeld:

graph_spec = builder.done()

GraphSpec

Grafiekspecificatie met gegevenspijplijn, schema en weergavemogelijkheden.

Constructor

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

Parameters:

• name (str): Grafieknaam
• context (ExecutionContext): Uitvoeringscontext
• graph_schema (GraphSchema): Grafiekschemadefinitie
• etl_pipeline (ETLPipeline, optioneel): Gegevenspijplijn voor graafvoorbereiding

Eigenschappen

nodes

def nodes() -> DataFrame

DataFrame-knooppunten ophalen (luie, in de cache). Bepaalt automatisch de bron van de gegevenspijplijn of lake-tabel.

Retourneert:

• DataFrame: Spark DataFrame met alle knooppunten

Verhoogt:

• ValueError: als er context ontbreekt of als DataFrames niet kunnen worden geladen

edges

def edges() -> DataFrame

Edges DataFrame ophalen (lui, in cache opgeslagen). Bepaalt automatisch de bron van de gegevenspijplijn of lake-tabel.

Retourneert:

• DataFrame: Spark DataFrame met alle randen

Verhoogt:

• ValueError: als er context ontbreekt of als DataFrames niet kunnen worden geladen

Methoden

build_graph_with_data

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

Voer de gegevenspijplijn uit en publiceer de grafiek. Roept intern aan Graph.build(self), past de geretourneerde Graphaan en retourneert een achterwaarts compatibele woordenlijst.

Retourneert:

• Dict[str, Any]: Woordenlijst met:
  • etl_result: Resultaten van gegevensvoorbereiding
  • api_result: Resultaten publiceren (indien geslaagd)
  • api_error: Fouttekenreeks (als publiceren is mislukt)
  • instance_name: Naam van graph-exemplaar
  • status: "published" of "prepared"

Voorbeeld:

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

get_schema

def get_schema() -> GraphSchema

Haal het grafiekschema op.

Retourneert:

• GraphSchema: Grafiekschemadefinitie

get_pipeline

def get_pipeline() -> Optional[ETLPipeline]

Haal de gegevenspijplijn op (Geen voor bestaande grafieken).

Retourneert:

• ETLPipeline of None: gegevenspijplijn indien beschikbaar

to_graphframe

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

Converteer de hele grafiek naar GraphFrame voor het uitvoeren van grafiekalgoritmen. Werkt alleen op lokale gegevens (uit de gegevenspijplijn of lake-tabel).

Parameters:

• column_mapping (Dict[str, str], optioneel): Aangepaste kolomtoewijzing met sleutels:
  • "id": Kolomnaam van hoekpunt-id
  • "source_id": Kolomnaam van edge-bron-id
  • "target_id": Kolomnaam van Edge-doel-id

Retourneert:

• GraphFrame: GraphFrame-object met alle hoekpunten en randen

Verhoogt:

• ValueError: Als ExecutionContext niet beschikbaar is

Voorbeeld:

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

show

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

Grafiekgegevens weergeven in verschillende indelingen.

Parameters:

• limit (int, default=100): maximum aantal knooppunten/randen dat moet worden weergegeven
• viz_format (str, default="visual"): uitvoerindeling
  • "table": Volledige DataFrame-tabellen (alle kolommen)
  • "visual": Interactieve grafiekvisualisatie
  • "all": Alle indelingen weergeven

Verhoogt:

• ValueError: Als de indeling niet een van de ondersteunde waarden is

Voorbeeld:

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

show_schema

def show_schema() -> None

Het grafiekschema weergeven als een interactieve grafiekvisualisatie.

Voorbeeld:

spec.show_schema()

Grafiek

Querybare grafiekexemplaren. Gemaakt via Graph.get() (bestaande grafiek) of Graph.build() (van een GraphSpec).

Constructor

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

Parameters:

• name (str): Grafieknaam
• context (ExecutionContext): Uitvoeringscontext
• spec (GraphSpec, optioneel): Bijgevoegde grafiekspecificatie (ingesteld door Graph.build())
• build_status (BuildStatus, optioneel): Metagegevens van buildresultaten (ingesteld door Graph.build())

Verhoogt:

• ValueError: Als ExecutionContext geen is

Statische methoden

get

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

Haal een grafiekexemplaren op uit een bestaande grafiek. De geretourneerde Graph heeft spec=None en build_status=None.

Parameters:

• name (str): Naam van graph-exemplaar
• context (ExecutionContext, optioneel): Uitvoeringscontext (standaard executionContext.default())

Retourneert:

• Graph: Graph-exemplaar

Verhoogt:

• ValueError: als de grafieknaam leeg is of als het grafiekexemplaren niet bestaat

Voorbeeld:

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

prepare

Graph.prepare(spec: GraphSpec) -> Graph

Voer de fase voor gegevensvoorbereiding uit voor een GraphSpeczonder te publiceren. Gebruik publish() daarna om de grafiek te registreren en er query's op uit te voeren.

Parameters:

• spec (GraphSpec): Grafiekspecificatie om voor te bereiden

Retourneert:

• Graph: Graph-exemplaar met spec gekoppelde en build_status.status == "prepared"

Verhoogt:

• ValueError: Als de specificatie geen gegevenspijplijn of uitvoeringscontext heeft
• RuntimeError: Als uitvoering van gegevenspijplijn mislukt

Voorbeeld:

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

Bouw een grafiek van een GraphSpec door gegevens voor te bereiden en te publiceren. Roept intern aan Graph.prepare(spec) en probeert graph.publish()vervolgens . In tegenstelling tot het afzonderlijk aanroepen van deze twee methoden, worden publicatiefouten opgevangen: de geretourneerde grafiek heeft build_status.status == "prepared" en build_status.api_error ingesteld in plaats van omhoog te gaan.

Parameters:

• spec (GraphSpec): Graph-specificatie om van te bouwen

Retourneert:

• Graph: Graph-exemplaar met spec gekoppelde en build_status ingevulde

Verhoogt:

• ValueError: Als de specificatie geen gegevenspijplijn of uitvoeringscontext heeft
• RuntimeError: Als uitvoering van gegevenspijplijn mislukt

Voorbeeld:

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

Eigenschappen

nodes

def nodes() -> Optional[DataFrame]

DataFrame voor knooppunten ophalen. Delegert naar self.spec.nodes wanneer een specificatie is gekoppeld; retourneert None anders.

edges

def edges() -> Optional[DataFrame]

Get edges DataFrame. Delegert naar self.spec.edges wanneer een specificatie is gekoppeld; retourneert None anders.

schema

def schema() -> Optional[GraphSchema]

Grafiekschema ophalen. Delegert naar self.spec.get_schema() wanneer een specificatie is gekoppeld; retourneert None anders.

Methoden

query

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

Voer een query uit op het grafiekexemplaren met behulp van GQL.

Parameters:

• query_string (str): Grafiekquerytekenreeks (GQL-taal)
• query_language (str, default="GQL"): Querytaal

Retourneert:

• QueryResult: Object met knooppunten, randen en metagegevens

Verhoogt:

• ValueError: Als ExecutionContext of Spark-sessie ontbreekt
• RuntimeError: als de initialisatie van de client of het uitvoeren van query's mislukt

Voorbeeld:

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

[! OPMERKING] reachability(query_input=ReachabilityQueryInput(...)) wordt nog steeds geaccepteerd, maar wordt verzonden DeprecationWarning en wordt verwijderd in een toekomstige versie.

Een bereikbaarheidsanalyse uitvoeren tussen bron- en doelknooppunten.

Parameters:

• source_property_value (str): waarde die overeenkomt met de broneigenschap (gevalideerd tijdens runtime. Moet worden opgegeven wanneer u niet gebruikmaakt van query_input)
• target_property_value (str): waarde die overeenkomt met de doeleigenschap (gevalideerd tijdens runtime. Moet worden opgegeven wanneer u niet gebruikmaakt van query_input)
• source_property (Optioneel[str]): eigenschapsnaam om bronknooppunten te filteren
• participating_source_node_labels (Optioneel[List[str]]): Knooppuntlabels te beschouwen als bronnen
• target_property (Optioneel[str]): eigenschapsnaam om doelknooppunten te filteren
• participating_target_node_labels (Optioneel[List[str]]): Knooppuntlabels te beschouwen als doelen
• participating_edge_labels (Optioneel[List[str]]): Edge-labels om door te bladeren
• is_directional (bool): of de randen richting hebben (standaard: True)
• min_hop_count (int): Minimale hops (standaard: 1)
• max_hop_count (int): Maximum aantal hops (standaard: 4)
• shortest_path (bool): retourneer alleen de kortste paden (standaard: False)
• max_results (int): Maximale resultaten (standaard: 500)

Verhoogt:

• ValueError: Als source_property_value of target_property_value ontbreekt, min_hop_count < 1, max_hop_count < min_hop_countof max_results < 1
• RuntimeError: als de initialisatie van de client of het uitvoeren van query's mislukt

Retourneert:

• QueryResult: Bevat de bereikbaarheidspaden

Voorbeeld:

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

k_hop

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

K-hopanalyse uitvoeren vanaf een bepaald bronknooppunt.

Parameters:

• Hetzelfde als reachability

Validatie:

• Ten minste één van source_property_value of target_property_value moet worden opgegeven

Verhoogt:

• ValueError: Als noch source_property_valuetarget_property_value wordt opgegeven, of als numerieke beperkingen worden geschonden (hetzelfde als reachability)
• RuntimeError: als de initialisatie van de client of het uitvoeren van query's mislukt

Retourneert:

• QueryResult: Bevat de k-hop resultaten

Voorbeeld:

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

Straalanalyse uitvoeren van bronknooppunt naar doelknooppunt.

Parameters:

• source_property_value (str): waarde ter identificatie van het bronknooppunt (gevalideerd tijdens runtime). Moet worden opgegeven wanneer u niet gebruikmaakt van query_input)
• target_property_value (str): waarde waarmee het doelknooppunt wordt geïdentificeerd (gevalideerd tijdens runtime. Moet worden opgegeven wanneer u niet gebruikmaakt van query_input)
• Andere parameters: hetzelfde als reachability

Verhoogt:

• ValueError: Als source_property_value of target_property_value ontbreekt of als numerieke beperkingen worden geschonden (hetzelfde als reachability)
• RuntimeError: als de initialisatie van de client of het uitvoeren van query's mislukt

Retourneert:

• QueryResult: Bevat de straalresultaten van de ontploffing

Voorbeeld:

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

Centraliteitsanalyse uitvoeren op de grafiek.

Parameters:

• participating_source_node_labels (Optioneel[List[str]]): Bronknooppuntlabels
• participating_target_node_labels (Optioneel[List[str]]): Doelknooppuntlabels
• participating_edge_labels (Optioneel[List[str]]): Edge-labels om door te bladeren
• threshold (int): minimale centraliteitsscore (standaard: 3); moet niet-negatief zijn
• centrality_type (CentralityType): CentralityType.Node of CentralityType.Edge (standaard: None, valt terug op CentralityType.Node)
• max_paths (int): het maximum aantal paden dat moet worden overwogen (standaard: 1000000; 0 = alle paden); moet niet-negatief zijn
• is_directional (bool): of de randen richting hebben (standaard: True)
• min_hop_count (int): minimale hops (standaard: 1); moeten ≥ 1 zijn
• max_hop_count (int): Maximum aantal hops (standaard: 4); moet ≥ min_hop_count
• shortest_path (bool): retourneer alleen de kortste paden (standaard: False)
• max_results (int): Maximale resultaten (standaard: 500); moeten ≥ 1 zijn

Verhoogt:

• ValueError: Als threshold < 0, max_paths < 0, min_hop_count < 1, max_hop_count < min_hop_countof max_results < 1
• RuntimeError: als de initialisatie van de client of het uitvoeren van query's mislukt

Retourneert:

• QueryResult: bevat de metrische gegevens voor centraliteit

Voorbeeld:

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

Een gerangschikte analyse uitvoeren op de grafiek.

Parameters:

• rank_property_name (str): eigenschapsnaam die moet worden gebruikt voor classificatie (gevalideerd tijdens runtime. Moet worden opgegeven wanneer u niet gebruikmaakt van query_input)
• threshold (int): alleen paden retourneren boven dit gewicht (standaard: 0); mogen niet-negatief zijn
• max_paths (int): het maximum aantal paden dat moet worden overwogen (standaard: 1000000; 0 = alle paden); moet niet-negatief zijn
• decay_factor (float): Rangschikking verval per stap; 2 betekent halvering (standaard: 1); moet niet-negatief zijn
• is_directional (bool): of de randen richting hebben (standaard: True)
• min_hop_count (int): minimale hops (standaard: 1); moeten ≥ 1 zijn
• max_hop_count (int): Maximum aantal hops (standaard: 4); moet ≥ min_hop_count
• shortest_path (bool): retourneer alleen de kortste paden (standaard: False)
• max_results (int): Maximale resultaten (standaard: 500); moeten ≥ 1 zijn

Verhoogt:

• ValueError: Als rank_property_name ontbreekt, threshold < 0, max_paths < 0, decay_factor < 0, min_hop_count < 1, max_hop_count < min_hop_count, of max_results < 1
• RuntimeError: als de initialisatie van de client of het uitvoeren van query's mislukt

Retourneert:

• QueryResult: Bevat de gerangschikte knooppunten/randen

Voorbeeld:

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

Volledige grafiek converteren naar GraphFrame. Maakt gebruik van spec-gegevens indien beschikbaar; leest anders uit lake-tabellen.

Parameters:

• column_mapping (Dict[str, str], optioneel): Aangepaste kolomtoewijzing

Retourneert:

• GraphFrame: GraphFrame-object met alle hoekpunten en randen

Voorbeeld:

gf = graph.to_graphframe()

show

def show() -> None

Grafiekgegevens weergeven. Delegert naar spec.show() voor uitgebreide weergave wanneer een specificatie is gekoppeld; anders wordt minimale informatie afgedrukt.

show_schema

def show_schema() -> None

Grafiekschema weergeven. Delegeren naar spec.show_schema() wanneer een specificatie is bijgevoegd; er wordt een bericht afgedrukt dat aangeeft dat er anders geen schema beschikbaar is.

publish (nieuw in v0.3.3)

def publish() -> Graph

Registreer de grafiek bij de API, zodat er query's kunnen worden uitgevoerd. Roep dit aan na Graph.prepare() (of op een willekeurige met Graph een specificatie) om het grafiekexemplaren te publiceren.

Retourneert:

• Graph: Zelf voor methode chaining

Verhoogt:

• ValueError: als er geen specificatie is gekoppeld of als de context ontbreekt
• RuntimeError: Als publiceren mislukt

Voorbeeld:

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

BuildStatus

Gegevensklasse die metagegevens van een Graph.build() bewerking bevat.

Velden

Bouwpaden

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)

Voorbeeld:

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

Opbouwfuncties voor knooppunten

NodeBuilderInitial

Initiële status voor knooppuntbouwer: alleen gegevensbronmethoden beschikbaar.

Constructor

NodeBuilderInitial(alias: str, graph_builder: GraphSpecBuilder)

Opmerking: Meestal gemaakt via GraphSpecBuilder.add_node(), niet rechtstreeks geïnstantieerd.

Methoden

from_table

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

Stel de tabel in als gegevensbron met intelligente databaseomzetting.

Parameters:

• table_name (str): naam van de tabel (vereist)
• database (str, optioneel): expliciete databasenaam (heeft voorrang op standaardcontext)

Retourneert:

• NodeBuilderSourceSet: Opbouwfunctie voor verdere configuratie

Verhoogt:

• ValueError: Als de tabel niet is gevonden of als er meerdere conflicterende tabellen zijn gevonden

Databaseomzettingsvolgorde:

1. Expliciete database parameter (hoogste prioriteit)
2. ExecutionContext.default_database
3. Zoeken in alle databases (met conflictdetectie)

Voorbeeld:

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

from_dataframe

def from_dataframe(dataframe: DataFrame) -> NodeBuilderSourceSet

Stel Spark DataFrame in als gegevensbron.

Parameters:

• dataframe (DataFrame): Spark DataFrame

Retourneert:

• NodeBuilderSourceSet: Opbouwfunctie voor verdere configuratie

Voorbeeld:

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

NodeBuilderSourceSet

Opbouwfunctie voor knooppunten nadat de gegevensbron is ingesteld: configuratiemethoden beschikbaar.

Constructor

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

Opmerking: Intern gemaakt door NodeBuilderInitial-bronmethoden.

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

Tijdsbereikfilter toepassen op de gegevensbron van het knooppunt.

Parameters:

• time_column (str): kolomnaam met tijdstempelgegevens (vereist)
• start_time (str of datetime, optioneel): Begindatum ('20-10-25', '2025-10-20', of datetime-object)
• end_time (str of datetime, optioneel): Einddatum (dezelfde notaties als start_time)
• lookback_hours (float, optioneel): uren om vanaf nu terug te kijken

Retourneert:

• NodeBuilderSourceSet: Zelf voor methode chaining

Verhoogt:

• ValueError: Als de tijdkolom niet is gevonden in het bronschema

Tijdsbereiklogica:

1. Als start_time en end_time opgegeven: gebruik deze rechtstreeks
2. Als lookback_hours opgegeven: end=now, start=now-lookback_hours
3. Als er niets is opgegeven: geen tijdfilter
4. Als begin/einde EN lookback_hours: begin/einde hebben voorrang

Voorbeeld:

# 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

Stel het knooppuntlabel in (wordt standaard ingesteld op alias als deze niet wordt aangeroepen).

Parameters:

• label (str): Knooppuntlabel

Retourneert:

• NodeBuilderSourceSet: Zelf voor methode chaining

Verhoogt:

• ValueError: Als label al is ingesteld

Voorbeeld:

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

with_columns

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

Kolommen configureren met de vereiste sleutel- en weergaveaanduiding.

Parameters:

• *columns (str): op te nemen kolomnamen (ten minste één vereist)
• key (str): Kolomnaam om als sleutel te markeren (vereist, moet in kolommen staan)
• display (str): Kolomnaam die moet worden gemarkeerd als weergavewaarde (vereist, moet in kolommen staan, kan hetzelfde zijn als de sleutel)

Retourneert:

• NodeBuilderSourceSet: Zelf voor methode chaining

Verhoogt:

• ValueError: Als validatie mislukt (dubbele kolommen, ontbrekende sleutel/weergave, enzovoort)

Opmerkingen:

• Eigenschappen worden automatisch gebouwd op basis van kolomtypen

• De kolom Tijdfilter wordt automatisch toegevoegd als deze is opgegeven

• Eigenschapstypen worden automatisch afgeleid uit het bronschema

• Zie Beperkingen

Voorbeeld:

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

Voltooi dit knooppunt en begin met het bouwen van een ander knooppunt.

Parameters:

• alias (str): alias voor het nieuwe knooppunt

Retourneert:

• NodeBuilderInitial: Nieuwe opbouwfunctie voor knooppunten

Voorbeeld:

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

Voltooi dit knooppunt en begin met het bouwen van een edge.

Parameters:

• alias (str): alias voor de rand

Retourneert:

• EdgeBuilderInitial: Nieuwe edge builder

Voorbeeld:

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

done

def done() -> GraphSpec

Voltooi dit knooppunt en voltooi de grafiekspecificatie.

Retourneert:

• GraphSpec: Grafiekspecificatie voltooien

Voorbeeld:

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

Edge Builders

EdgeBuilderInitial

Initiële status voor Edge Builder: alleen gegevensbronmethoden beschikbaar.

Constructor

EdgeBuilderInitial(alias: str, graph_builder: GraphSpecBuilder)

Opmerking: Meestal gemaakt via GraphSpecBuilder.add_edge(), niet rechtstreeks geïnstantieerd.

Methoden

from_table

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

Stel de tabel in als gegevensbron met intelligente databaseomzetting.

Parameters:

• table_name (str): naam van de tabel (vereist)
• database (str, optioneel): Expliciete databasenaam

Retourneert:

• EdgeBuilderSourceSet: Opbouwfunctie voor verdere configuratie

Verhoogt:

• ValueError: Als de tabel niet is gevonden of als er meerdere conflicterende tabellen zijn gevonden

Voorbeeld:

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

from_dataframe

def from_dataframe(dataframe: DataFrame) -> EdgeBuilderSourceSet

Stel Spark DataFrame in als gegevensbron.

Parameters:

• dataframe (DataFrame): Spark DataFrame

Retourneert:

• EdgeBuilderSourceSet: Opbouwfunctie voor verdere configuratie

Voorbeeld:

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

EdgeBuilderSourceSet

Edge Builder nadat de gegevensbron is ingesteld: configuratiemethoden beschikbaar.

Constructor

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

Opmerking: Intern gemaakt door EdgeBuilderInitial-bronmethoden.

Methoden

with_label

def with_label(label: str) -> EdgeBuilderSourceSet

Stel het type/label van de randrelatie in (wordt standaard ingesteld op alias als deze niet wordt aangeroepen).

Parameters:

• label (str): Edge-label

Retourneert:

• EdgeBuilderSourceSet: Zelf voor methode chaining

Verhoogt:

• ValueError: Als label al is ingesteld

Voorbeeld:

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

edge_label

def edge_label(label: str) -> EdgeBuilderSourceSet

Stel het type/label van de randrelatie in (wordt standaard ingesteld op alias als deze niet wordt aangeroepen).

Parameters:

• label (str): Edge-label

Retourneert:

• EdgeBuilderSourceSet: Zelf voor methode chaining

Verhoogt:

• ValueError: Als label al is ingesteld

Voorbeeld:

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

source

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

Stel bronknooppunt in met id-kolom en label.

Parameters:

• id_column (str): Kolomnaam met bronknooppunt-id
• node_type (str): Bronknooppuntlabel

Retourneert:

• EdgeBuilderSourceSet: Zelf voor methode chaining

Verhoogt:

• ValueError: Als de bron al is ingesteld

Voorbeeld:

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

Doelknooppunt instellen met id-kolom en label.

Parameters:

• id_column (str): kolomnaam met doelknooppunt-id
• node_type (str): doelknooppuntlabel

Retourneert:

• EdgeBuilderSourceSet: Zelf voor methode chaining

Verhoogt:

• ValueError: Als het doel al is ingesteld

Voorbeeld:

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

Tijdsbereikfilters toepassen op de gegevensbron van de edge.

Parameters:

• time_column (str): kolomnaam met tijdstempelgegevens (vereist)
• start_time (str of datetime, optioneel): Begindatum
• end_time (str of datetime, optioneel): Einddatum
• lookback_hours (float, optioneel): uren om vanaf nu terug te kijken

Retourneert:

• EdgeBuilderSourceSet: Zelf voor methode chaining

Verhoogt:

• ValueError: Als de tijdkolom niet is gevonden in het bronschema

Voorbeeld:

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

Kolommen configureren met de vereiste sleutel- en weergaveaanduiding.

Parameters:

• *columns (str): op te nemen kolomnamen (ten minste één vereist)
• key (str): Kolomnaam om als sleutel te markeren (vereist, moet in kolommen staan)
• display (str): Kolomnaam die moet worden gemarkeerd als weergavewaarde (vereist, moet in kolommen staan)

Retourneert:

• EdgeBuilderSourceSet: Zelf voor methode chaining

Verhoogt:

• ValueError: Als de validatie mislukt

Voorbeeld:

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

Voltooi deze rand en begin met het bouwen van een knooppunt.

Parameters:

• alias (str): alias voor het nieuwe knooppunt

Retourneert:

• NodeBuilderInitial: Nieuwe opbouwfunctie voor knooppunten

add_edge

def add_edge(alias: str) -> EdgeBuilderInitial

Voltooi deze rand en begin met het bouwen van een andere rand.

Parameters:

• alias (str): alias voor de nieuwe rand

Retourneert:

• EdgeBuilderInitial: Nieuwe edge builder

Voorbeeld:

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

Voltooi deze rand en voltooi de grafiekspecificatie.

Retourneert:

• GraphSpec: Grafiekspecificatie voltooien

Schemaklassen

GraphDefinitionReference

Verwijzing naar een grafiekdefinitie met naam en versie.

Constructor

GraphDefinitionReference(
    fully_qualified_name: str,
    version: str
)

Parameters:

• fully_qualified_name (str): Volledig gekwalificeerde naam van de grafiek waarnaar wordt verwezen
• version (str): versie van de grafiek waarnaar wordt verwezen

Verhoogt:

• ValueError: Als fully_qualified_name of versie leeg is

Methoden

to_dict

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

Serialiseren naar woordenlijst.

Retourneert:

• Dict[str, Any]: Geserialiseerde verwijzing

Eigenschap

Eigenschapsdefinitie met typeveilige interface.

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
)

Parameters:

• name (str): Eigenschapsnaam
• property_type (PropertyType): Gegevenstype eigenschap
• is_non_null (bool, default=False): of de eigenschap vereist is
• description (str, default=""): Beschrijving van eigenschap
• is_key (bool, default=False): of de eigenschap een sleutel is
• is_display_value (bool, default=False): of de eigenschap weergavewaarde is
• is_internal (bool, default=False): of de eigenschap intern is

Verhoogt:

• ValueError: als de naam leeg is of de validatie mislukt

Klassemethoden

key

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

Maak een sleuteleigenschap met algemene instellingen (is_key=True, is_display_value=True).

display

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

Maak een eigenschap weergavewaarde (is_display_value=Waar).

Methoden

describe

def describe(text: str) -> Property

Voeg de beschrijving vloeiend toe.

Parameters:

• text (str): Beschrijvingstekst

Retourneert:

• Property: Zelf voor methode chaining

to_dict

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

De eigenschap serialiseren naar woordenlijst met @-voorvoegsel annotatiesleutels.

Retourneert:

• Dict[str, Any]: Geserialiseerde eigenschap

to_gql

def to_gql() -> str

GQL-eigenschapsdefinitie genereren.

Retourneert:

• str: GQL-tekenreeksweergave

EdgeNode

Verwijzing naar knooppunten die worden gebruikt in edge-definities.

Constructor

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

Parameters:

• alias (str, optioneel): Knooppuntalias (automatisch ingesteld op eerste label als geen of leeg)
• labels (Lijst[str]): Knooppuntlabels (ten minste één vereist)

Verhoogt:

• ValueError: Als de lijst met labels leeg is
• TypeError: Als labels geen tekenreeksen zijn

Automatische mutatie:

• Als alias Geen of leeg is, wordt deze ingesteld op het eerste label

Methoden

to_dict

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

Serialiseren naar woordenlijst.

Retourneert:

• Dict[str, Any]: Verwijzing naar geserialiseerde edge-knooppunten

Knooppunt

Knooppuntdefinitie met typeveilige interface.

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
)

Parameters:

• alias (str, default=""): Knooppuntalias (automatisch ingesteld op eerste label indien leeg)
• labels (Lijst[str]): Knooppuntlabels (ten minste één vereist)
• implies_labels (List[str], default=[]): Impliciete labels
• properties (List[Property], default=[]): Knooppunteigenschappen
• description (str, default=""): Knooppuntbeschrijving
• entity_group (str, default=""): Naam van entiteitsgroep
• dynamic_labels (bool, default=False): of het knooppunt dynamische labels heeft
• abstract_edge_aliases (bool, default=False): of knooppunt abstracte edge-aliassen gebruikt

Verhoogt:

• ValueError: Als validatie mislukt (geen labels, geen sleuteleigenschap, geen weergave-eigenschap, enzovoort)

Automatische mutatie:

• Als de alias leeg is, wordt deze ingesteld op het eerste label
• Als entity_group leeg is, wordt deze ingesteld op het primaire label

Methoden

get_primary_label

def get_primary_label() -> Optional[str]

Haal het primaire (eerste) label op.

Retourneert:

• str of None: Primair label of Geen als er geen labels zijn

get_entity_group_name

def get_entity_group_name() -> str

Haal de naam van de entiteitsgroep op of terugval naar het primaire label.

Retourneert:

• str: Naam van entiteitsgroep

get_primary_key_property_name

def get_primary_key_property_name() -> Optional[str]

Haal de naam op van de primaire sleuteleigenschap.

Retourneert:

• str of None: naam van primaire sleuteleigenschap

get_properties

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

Eigenschappen ophalen als een woordenlijst voor eenvoudige toegang.

Retourneert:

• Dict[str, Property]: Eigenschappen op naam

get_property

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

Een specifieke eigenschap ophalen op naam.

Parameters:

• name (str): Eigenschapsnaam

Retourneert:

• Property of None: eigenschap indien gevonden

add_property

def add_property(prop: Property) -> None

Voeg een eigenschap toe aan dit knooppunt.

Parameters:

• prop (Eigenschap): eigenschap om toe te voegen

Verhoogt:

• ValueError: Als de eigenschapsnaam dubbel is

is_dynamically_labeled

def is_dynamically_labeled() -> bool

Controleer of het knooppunt dynamische labels heeft.

Retourneert:

• bool: Waar als dynamische labels zijn ingeschakeld

is_abstract_edge_node_aliases

def is_abstract_edge_node_aliases() -> bool

Controleer of het knooppunt abstracte edge-knooppuntaliassen gebruikt.

Retourneert:

• bool: Waar als abstracte edge-aliassen zijn ingeschakeld

describe

def describe(text: str) -> Node

Voeg de beschrijving vloeiend toe.

Parameters:

• text (str): Beschrijvingstekst

Retourneert:

• Node: Zelf voor methode chaining

to_dict

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

Serialiseer knooppunt naar woordenlijst.

Retourneert:

• Dict[str, Any]: Geserialiseerd knooppunt

to_gql

def to_gql() -> str

Genereer GQL-knooppuntdefinitie.

Retourneert:

• str: GQL-tekenreeksweergave

Verhoogt:

• ValueError: Als het knooppunt geen vereiste velden voor GQL bevat

Klassemethoden

create

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

Maak een knooppunt met alle vereiste velden.

Parameters:

• alias (str): Knooppuntalias
• labels (Lijst[str]): Knooppuntlabels
• properties (List[Property]): Knooppunteigenschappen
• description (str, default=""): Knooppuntbeschrijving
• entity_group (str, default=""): Naam van entiteitsgroep

Retourneert:

• Node: Nieuw knooppuntexemplaren

Rand

Edge-definitie met typeveilige interface.

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
)

Parameters:

• relationship_type (str): Edge-relatietype (bijvoorbeeld 'VOLGT', 'EIGEN')
• source_node_label (str): Bronknooppuntlabel
• target_node_label (str): doelknooppuntlabel
• direction (EdgeDirection, default=DIRECTED_RIGHT): Edge direction
• properties (List[Property], default=[]): Edge-eigenschappen
• description (str, default=""): Edge-beschrijving
• entity_group (str, default=""): Naam van entiteitsgroep
• dynamic_type (bool, default=False): of edge dynamisch type heeft

Verhoogt:

• ValueError: Als de validatie mislukt

Automatische mutatie:

• labels lijst wordt automatisch ingevuld met [relationship_type]
• Als entity_group leeg is, wordt deze ingesteld op relationship_type

Eigenschappen

edge_type

def edge_type() -> str

Achterwaartse compatibiliteitsalias voor relationship_type.

Retourneert:

• str: Relatietype

Methoden

get_entity_group_name

def get_entity_group_name() -> str

Haal de naam van de entiteitsgroep op of terugval op relatietype.

Retourneert:

• str: Naam van entiteitsgroep

is_dynamic_type

def is_dynamic_type() -> bool

Controleer of edge een dynamisch type heeft.

Retourneert:

• bool: Waar als dynamisch type

add_property

def add_property(edge_property: Property) -> None

Voeg een eigenschap toe aan deze rand.

Parameters:

• edge_property (Eigenschap): eigenschap om toe te voegen

describe

def describe(text: str) -> Edge

Voeg de beschrijving vloeiend toe.

Parameters:

• text (str): Beschrijvingstekst

Retourneert:

• Edge: Zelf voor methode chaining

to_dict

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

Edge serialiseren naar woordenlijst.

Retourneert:

• Dict[str, Any]: Geserialiseerde rand

to_gql

def to_gql() -> str

Genereer GQL edge-definitie.

Retourneert:

• str: GQL-tekenreeksweergave

Klassemethoden

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

Maak een rand met alle vereiste velden.

Parameters:

• relationship_type (str): Type edge-relatie
• source_node_label (str): Bronknooppuntlabel
• target_node_label (str): doelknooppuntlabel
• properties (List[Property], optioneel): Edge-eigenschappen
• description (str, default=""): Edge-beschrijving
• entity_group (str, default=""): Naam van entiteitsgroep

Retourneert:

• Edge: Nieuw edge-exemplaar

GraphSchema

Grafiekschemadefinitie met typeveilige interface.

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

Parameters:

• name (str): Grafiekschemanaam
• nodes (List[Node], default=[]): Knooppuntdefinities
• edges (List[Edge], default=[]): Edge-definities
• base_graphs (List[GraphSchema], default=[]): Basisgrafiekschema's
• description (str, default=""): Schemabeschrijving
• version (str, default="1.0"): Schemaversie
• fully_qualified_name (str, default=""): Volledig gekwalificeerde naam
• namespace (str, default=""): naamruimte

Verhoogt:

• ValueError: als de validatie mislukt (dubbele aliassen, randen verwijzen naar niet-bestaande knooppunten, enzovoort)

Methoden

get_fully_qualified_name

def get_fully_qualified_name() -> str

Volledig gekwalificeerde naam ophalen.

Retourneert:

• str: Volledig gekwalificeerde naam

get_namespace

def get_namespace() -> str

Haal de naamruimte op van volledig gekwalificeerde naam of retourneer standaard.

Retourneert:

• str:Naamruimte

get_version

def get_version() -> str

Versie ophalen.

Retourneert:

• str: Versietekenreeks

get_node

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

Knooppunt ophalen op label of alias.

Parameters:

• label_or_alias (str): Knooppuntlabel of alias

Retourneert:

• Node of None: Knooppunt indien gevonden

get_edge

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

Edge ophalen op naam/type.

Parameters:

• name (str): Type edge-relatie

Retourneert:

• Edge of None: Edge indien gevonden

add_node

def add_node(node: Node) -> None

Voeg een knooppunt toe aan deze grafiek.

Parameters:

• node (Knooppunt): toe te voegen knooppunt

Verhoogt:

• ValueError: Als de knooppuntalias dubbel is

add_edge

def add_edge(edge: Edge) -> None

Voeg een rand toe aan deze grafiek.

Parameters:

• edge (Edge): Edge om toe te voegen

Verhoogt:

• ValueError: Als het randtype dubbel is

include_graph

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

Voeg een grafiektoevoeging toe (Fluent API).

Parameters:

• fully_qualified_name (str): Volledig gekwalificeerde naam van grafiek om op te nemen
• version (str): versie van grafiek die moet worden opgenomen

Retourneert:

• GraphSchema: Zelf voor methode chaining

get_included_graph_references

def get_included_graph_references() -> List[GraphDefinitionReference]

Lijst met opgenomen grafiekverwijzingen ophalen.

Retourneert:

• List[GraphDefinitionReference]: Lijst met grafiekdefinitieverwijzingen

describe

def describe(text: str) -> GraphSchema

Voeg de beschrijving vloeiend toe.

Parameters:

• text (str): Beschrijvingstekst

Retourneert:

• GraphSchema: Zelf voor methode chaining

to_dict

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

Schema serialiseren naar woordenlijst.

Retourneert:

• Dict[str, Any]: Geserialiseerd schema

to_json

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

JSON-weergave genereren.

Parameters:

• indent (int, default=2): JSON-inspringingsniveau

Retourneert:

• str: JSON-tekenreeks

to_gql

def to_gql() -> str

GQL-schemadefinitie genereren.

Retourneert:

• str: GQL-tekenreeksweergave

Klassemethoden

create

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

Maak een grafiekschema met alle vereiste velden.

Parameters:

• name (str): Grafiekschemanaam
• nodes (Lijst[Knooppunt], optioneel): Knooppuntdefinities
• edges (List[Edge], optioneel): Edge-definities
• description (str, default=""): Schemabeschrijving
• version (str, default="1.0"): Schemaversie

Retourneert:

• GraphSchema: Nieuw grafiekschema-exemplaar

Queryinvoerklassen

Gegevensklassen die invoerparameters vertegenwoordigen voor vooraf gedefinieerde grafiekquery's.

QueryInputBase

Basisklasse voor alle queryinvoerparameters.

Methoden

to_json_payload

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

Converteer de invoerparameters naar een woordenlijst voor API-verzending.

Retourneert:

• Dict[str, Any]: Woordenlijstweergave van de invoerparameters

validate

def validate() -> None

Valideer de invoerparameters.

Verhoogt:

• ValueError: Als de invoerparameters ongeldig zijn

ReachabilityQueryInput

Invoerparameters voor een bereikbaarheidsquery tussen bron- en doelknooppunten. Neemt over van ReachabilityQueryInputBase waarvan wordt overgenomen van QueryInputBase.

Velden

Validatie:

• source_property_value is vereist
• target_property_value is vereist

Voorbeeld:

# 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

Invoerparameters voor een k-hopquery van een bepaald bronknooppunt. Neemt over van ReachabilityQueryInputBase.

Neemt alle velden over van ReachabilityQueryInput.

Validatie:

• Ten minste één van source_property_value of target_property_value moet worden opgegeven

Voorbeeld:

# 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

Invoerparameters voor een straalquery van bron naar doelknooppunten. Neemt over van ReachabilityQueryInputBase.

Neemt alle velden over van ReachabilityQueryInput, met de volgende vereiste velden:

Validatie:

• source_property_value is vereist
• target_property_value is vereist

Voorbeeld:

# 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

Invoerparameters voor een centrale analysequery. Neemt over van QueryInputBase.

CentralityType Enum

Velden

Voorbeeld:

# 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

Invoerparameters voor een geclassificeerde analysequery. Neemt over van QueryInputBase.

Velden

Voorbeeld:

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

Queryresultaten

QueryResult

Resultaat van een grafiekquery met luie DataFrame-toegang.

Constructor

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

Parameters:

• raw_response (Dict[str, Any]): Onbewerkte API-antwoordwoordenlijst
• graph (Graph): verwijzing naar bovenliggende Graph

Opmerking: Meestal gemaakt door Graph.query(), niet rechtstreeks geïnstantieerd.

Methoden

to_dataframe

def to_dataframe() -> DataFrame

Converteert het queryresultaat naar een Spark DataFrame.

Retourneert:

• DataFrame: Queryresultaat als Spark DataFrame

Verhoogt:

• ValueError: Als de conversie mislukt

Voorbeeld:

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

get_raw_data

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

Haal de sectie RawData op uit het antwoord.

Retourneert:

• Dict[str, Any]: Woordenlijst met onbewerkte metagegevens of leeg dicteren als deze niet aanwezig is

Voorbeeld:

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

show

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

Queryresultaten weergeven in verschillende indelingen.

Parameters:

• format (str, default="visual"): uitvoerindeling
  • "table": Volledige DataFrame-tabellen (alle kolommen)
  • "visual": Interactieve grafiekvisualisatie met VSC-invoegtoepassing
  • "all": Alle indelingen weergeven

Verhoogt:

• ValueError: Als de indeling niet een van de ondersteunde waarden is

Voorbeeld:

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

Volledig voorbeeld (aanbevolen : v0.3+)

# 0. Imports
from sentinel_graph import GraphSpecBuilder, Graph

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

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

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

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

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

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

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

Notities over ontwerppatronen

Fluent-API

Alle opbouwfuncties ondersteunen methodekoppeling voor leesbare, declaratieve grafiekdefinities:

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

Union-schema's

Meerdere randen met dezelfde alias worden automatisch samengevoegd met samengevoegde eigenschappen:

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

Automatische configuratie

Veel velden hebben verstandige standaardwaarden:

• Knooppunt-/edge-labels zijn standaard ingesteld op hun aliassen
• Eigenschappen worden automatisch afgeleid uit bronschema's
• Entiteitsgroepen zijn standaard ingesteld op primaire labels/relatietypen

Luie evaluatie

DataFrames en resources worden lui geladen en in de cache opgeslagen:

• graph_spec.nodes en graph_spec.edges worden geladen bij de eerste toegang
• Queryresultaten maken alleen DataFrames wanneer daarom wordt gevraagd

Verwante onderwerpen

• Overzicht van aangepaste grafieken
• Aangepaste grafieken maken