Informations de référence sur l’API Graph Builder (préversion)

La classe sentinel_graph permet d’interagir avec le graphe Microsoft Sentinel, ce qui vous permet de définir votre schéma de graphe, de transformer les données du lac de données Microsoft Sentinel en nœuds et arêtes, de publier un graphe, d’interroger un graphique et d’exécuter des algorithmes de graphe avancés. Cette classe est conçue pour fonctionner avec les sessions Spark dans les notebooks Jupyter s’exécutant sur Microsoft Sentinel calcul Spark.

GraphSpecBuilder

La classe GraphSpecBuilder fournit un générateur Fluent pour créer des spécifications de graphe avec des pipelines de données et l’intégration de schéma.

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

# Recommended
from sentinel_graph import GraphSpecBuilder

Constructeur

GraphSpecBuilder(context: ExecutionContext)

Paramètres :

• context (ExecutionContext) : contexte d’exécution contenant la session spark et la configuration

Soulève:

• ValueError: si le contexte a la valeur None ou si le nom du graphique ne peut pas être déterminé

Méthodes statiques

start

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

Définissez un nouveau générateur de graphes Fluent.

Paramètres :

• context(ExecutionContext, facultatif) : ExecutionContext instance. Si aucun, utilise le contexte par défaut.

Retourne:

• GraphSpecBuilder: Nouveau instance de générateur

Exemple :

builder = GraphSpecBuilder.start(context=context)

Méthodes d’instance

add_node

def add_node(alias: str) -> NodeBuilderInitial

Commencez à créer une définition de nœud.

Paramètres :

• alias (str) : identificateur unique de ce nœud dans le graphique

Retourne:

• NodeBuilderInitial: Générateur de nœuds dans l’état initial

Exemple :

builder.add_node("user")

add_edge

def add_edge(alias: str) -> EdgeBuilderInitial

Commencez à créer une définition d’arête.

Paramètres :

• alias (str) : identificateur de cette arête dans le graphe (peut être partagé entre plusieurs arêtes)

Retourne:

• EdgeBuilderInitial: Générateur Edge dans l’état initial

Exemple :

builder.add_edge("accessed")

done

def done() -> GraphSpec

Finalisez la spécification du graphique et retournez l’instance GraphSpec.

Retourne:

• GraphSpec: Spécification complète du graphique avec le pipeline de données et le schéma

Soulève:

• ValueError: si le graphique n’a pas de nœuds ou d’arêtes, ou si la validation échoue

Exemple :

graph_spec = builder.done()

GraphSpec

Spécification de graphique avec des fonctionnalités de pipeline de données, de schéma et d’affichage.

Constructeur

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

Paramètres :

• name (str) : nom du graphique
• context (ExecutionContext) : contexte d’exécution
• graph_schema (GraphSchema) : définition de schéma de graphe
• etl_pipeline (ETLPipeline, facultatif) : pipeline de données pour la préparation du graphique

Propriétés

nodes

def nodes() -> DataFrame

Obtenir le DataFrame des nœuds (différé, mis en cache). Détermine automatiquement la source à partir du pipeline de données ou de la table de lac.

Retourne:

• DataFrame: DataFrame Spark contenant tous les nœuds

Soulève:

• ValueError: si le contexte est manquant ou si les DataFrames ne peuvent pas être chargés

edges

def edges() -> DataFrame

Obtenir le DataFrame edges (différé, mis en cache). Détermine automatiquement la source à partir du pipeline de données ou de la table de lac.

Retourne:

• DataFrame: DataFrame Spark contenant tous les bords

Soulève:

• ValueError: si le contexte est manquant ou si les DataFrames ne peuvent pas être chargés

Méthodes

build_graph_with_data

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

Exécutez le pipeline de données et publiez le graphique. Appelle en Graph.build(self)interne , stash le retourné Graphet retourne un dictionnaire à compatibilité descendante.

Retourne:

• Dict[str, Any]: Dictionnaire contenant :
  • etl_result: résultats de la préparation des données
  • api_result: Publier les résultats (en cas de réussite)
  • api_error: Chaîne d’erreur (si la publication a échoué)
  • instance_name: Nom du instance graph
  • status: "published" ou "prepared"

Exemple :

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

get_schema

def get_schema() -> GraphSchema

Obtient le schéma de graphe.

Retourne:

• GraphSchema: définition du schéma de graphe

get_pipeline

def get_pipeline() -> Optional[ETLPipeline]

Obtenir le pipeline de données (Aucun pour les graphiques existants).

Retourne:

• ETLPipeline ou None: pipeline de données si disponible

to_graphframe

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

Convertissez l’intégralité du graphique en GraphFrame pour exécuter des algorithmes de graphe. Fonctionne uniquement sur des données locales (à partir d’un pipeline de données ou d’une table de lac).

Paramètres :

• column_mapping (Dict[str, str], facultatif) : mappage de colonnes personnalisé avec des clés :
  • "id": Nom de colonne ID de vertex
  • "source_id": Nom de colonne d’ID source Edge
  • "target_id": Nom de la colonne ID cible Edge

Retourne:

• GraphFrame: objet GraphFrame avec tous les sommets et bords

Soulève:

• ValueError: si ExecutionContext n’est pas disponible

Exemple :

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

show

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

Afficher des données de graphique dans différents formats.

Paramètres :

• limit (int, default=100) : nombre maximal de nœuds/arêtes à afficher
• viz_format (str, default="visual ») : Format de sortie
  • "table": tables de DataFrame complètes (toutes les colonnes)
  • "visual": visualisation graphique interactive
  • "all": Afficher tous les formats

Soulève:

• ValueError: si format n’est pas l’une des valeurs prises en charge

Exemple :

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

show_schema

def show_schema() -> None

Afficher le schéma de graphe sous forme de visualisation graphique interactive.

Exemple :

spec.show_schema()

Graph

Instance de graphique interrogeable. Créé via Graph.get() (graphe existant) ou Graph.build() (à partir d’un GraphSpec).

Constructeur

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

Paramètres :

• name (str) : nom du graphique
• context (ExecutionContext) : contexte d’exécution
• spec (GraphSpec, facultatif) : spécification de graphe jointe (définie par Graph.build())
• build_status (BuildStatus, facultatif) : métadonnées de résultat de build (définies par Graph.build())

Soulève:

• ValueError: si ExecutionContext a la valeur None

Méthodes statiques

get

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

Obtenir un instance de graphique à partir d’un graphique existant. Le retourné Graph a spec=None et build_status=None.

Paramètres :

• name(str) : nom du instance du graphique
• context (ExecutionContext, facultatif) : contexte d’exécution (valeur par défaut ExecutionContext.default())

Retourne:

• Graph: instance graphes

Soulève:

• ValueError: si le nom du graphique est vide ou si le instance de graphique n’existe pas

Exemple :

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

prepare

Graph.prepare(spec: GraphSpec) -> Graph

Exécutez la phase de préparation des données pour un GraphSpecsans publication. Utilisez publish() ensuite pour inscrire le graphe et le rendre interrogeable.

Paramètres :

• spec (GraphSpec) : spécification de graphe à préparer

Retourne:

• Graph: graphique instance avec spec attachés etbuild_status.status == "prepared"

Soulève:

• ValueError: si la spécification n’a pas de pipeline de données ou aucun contexte d’exécution
• RuntimeError: si l’exécution du pipeline de données échoue

Exemple :

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

Créez un graphique à partir d’un GraphSpec en préparant les données et en les publiant. Appelle en Graph.prepare(spec) interne, puis tente graph.publish(). Contrairement à l’appel de ces deux méthodes séparément, les échecs de publication sont interceptés : le graphique retourné a build_status.status == "prepared" et build_status.api_error défini au lieu de déclencher.

Paramètres :

• spec (GraphSpec) : spécification de graphe à partir de

Retourne:

• Graph: instance de graphique avec spec attaché et build_status rempli

Soulève:

• ValueError: si la spécification n’a pas de pipeline de données ou aucun contexte d’exécution
• RuntimeError: si l’exécution du pipeline de données échoue

Exemple :

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

Propriétés

nodes

def nodes() -> Optional[DataFrame]

Obtenir le DataFrame des nœuds. Délègue à self.spec.nodes lorsqu’une spécification est attachée ; retourne None dans le cas contraire.

edges

def edges() -> Optional[DataFrame]

Obtenir le DataFrame des arêtes. Délègue à self.spec.edges lorsqu’une spécification est attachée ; retourne None dans le cas contraire.

schema

def schema() -> Optional[GraphSchema]

Obtenir le schéma de graphe. Délègue à self.spec.get_schema() lorsqu’une spécification est attachée ; retourne None dans le cas contraire.

Méthodes

query

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

Exécutez une requête sur le graphique instance à l’aide de GQL.

Paramètres :

• query_string (str) : chaîne de requête de graphe (langage GQL)
• query_language (str, default="GQL ») : Langage de requête

Retourne:

• QueryResult: objet contenant des nœuds, des arêtes et des métadonnées

Soulève:

• ValueError: si la session ExecutionContext ou Spark est manquante
• RuntimeError: si l’initialisation du client ou l’exécution de requête échoue

Exemple :

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

[! REMARQUE] reachability(query_input=ReachabilityQueryInput(...)) est toujours accepté, mais émet DeprecationWarning et sera supprimé dans une version ultérieure.

Effectuer une analyse d’accessibilité entre les nœuds source et cible.

Paramètres :

• source_property_value (str) : valeur à mettre en correspondance pour la propriété source (validée au moment de l’exécution. Doit être fourni lorsque vous n’utilisez query_inputpas )
• target_property_value (str) : valeur à mettre en correspondance pour la propriété cible (validée lors de l’exécution. Doit être fourni lorsque vous n’utilisez query_inputpas )
• source_property (Facultatif[str]) : nom de propriété pour filtrer les nœuds sources
• participating_source_node_labels (Facultatif[List[str]]) : étiquettes de nœud à considérer comme sources
• target_property (Facultatif[str]) : nom de propriété pour filtrer les nœuds cibles
• participating_target_node_labels (Facultatif[List[str]]) : étiquettes de nœud à prendre en compte comme cibles
• participating_edge_labels (Facultatif[List[str]]) : étiquettes d’arêtes à parcourir
• is_directional (bool) : indique si les arêtes sont directionnelles (valeur par défaut : True)
• min_hop_count (int) : tronçons minimaux (valeur par défaut : 1)
• max_hop_count (int) : tronçons maximum (valeur par défaut : 4)
• shortest_path (bool) : retourne uniquement les chemins d’accès les plus courts (par défaut : False)
• max_results (int) : résultats maximum (par défaut : 500)

Soulève:

• ValueError: si source_property_value ou target_property_value est manquant, min_hop_count < 1, max_hop_count < min_hop_countou max_results < 1
• RuntimeError: si l’initialisation du client ou l’exécution de requête échoue

Retourne:

• QueryResult: contenant les chemins d’accès

Exemple :

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

Effectuez une analyse k-hop à partir d’un nœud source donné.

Paramètres :

• Identique à reachability

Validation:

• Au moins un de source_property_value ou target_property_value doit être fourni

Soulève:

• ValueError: si ni n’est source_property_valuetarget_property_value fourni, ou si les contraintes numériques sont violées (identiques à reachability)
• RuntimeError: si l’initialisation du client ou l’exécution de requête échoue

Retourne:

• QueryResult: contenant les résultats du k-hop

Exemple :

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

Effectuez une analyse du rayon d’explosion du nœud source au nœud cible.

Paramètres :

• source_property_value (str) : valeur identifiant le nœud source (validé lors de l’exécution. Doit être fourni lorsque vous n’utilisez query_inputpas )
• target_property_value (str) : valeur identifiant le nœud cible (validé au moment de l’exécution. Doit être fourni lorsque vous n’utilisez query_inputpas )
• Autres paramètres : identiques à reachability

Soulève:

• ValueError: si source_property_value ou target_property_value est manquant, ou si les contraintes numériques sont violées (identiques à reachability)
• RuntimeError: si l’initialisation du client ou l’exécution de requête échoue

Retourne:

• QueryResult: contenant les résultats du rayon d’explosion

Exemple :

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

Effectuez une analyse de centralité sur le graphique.

Paramètres :

• participating_source_node_labels (Facultatif[List[str]]) : étiquettes de nœud source
• participating_target_node_labels (Facultatif[List[str]]) : étiquettes de nœud cible
• participating_edge_labels (Facultatif[List[str]]) : étiquettes d’arêtes à parcourir
• threshold (int) : score de centralité minimal (valeur par défaut : 3) ; doit être non négatif
• centrality_type (CentralityType) : CentralityType.Node ou CentralityType.Edge (valeur par défaut : None, revient à CentralityType.Node)
• max_paths (int) : nombre maximal de chemins à prendre en compte (par défaut : 1000000; 0 = tous les chemins) ; doit être non négatif
• is_directional (bool) : indique si les arêtes sont directionnelles (valeur par défaut : True)
• min_hop_count (int) : tronçons minimaux (valeur par défaut : 1) ; doivent être ≥ 1
• max_hop_count (int) : le nombre maximal de sauts (valeur par défaut : 4) doit être ≥ min_hop_count
• shortest_path (bool) : retourne uniquement les chemins d’accès les plus courts (par défaut : False)
• max_results (int) : le nombre maximal de résultats (par défaut : 500) doit être ≥ 1

Soulève:

• ValueError: Si threshold < 0, max_paths < 0, min_hop_count < 1, max_hop_count < min_hop_countou max_results < 1
• RuntimeError: si l’initialisation du client ou l’exécution de requête échoue

Retourne:

• QueryResult: contenant les métriques de centralité

Exemple :

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

Effectuez une analyse classée sur le graphique.

Paramètres :

• rank_property_name (str) : nom de propriété à utiliser pour le classement (validé au moment de l’exécution. Doit être fourni lorsque vous n’utilisez query_inputpas )
• threshold (int) : seuls les chemins d’accès retournés au-dessus de cette pondération (valeur par défaut : 0) doivent être non négatifs
• max_paths (int) : nombre maximal de chemins à prendre en compte (par défaut : 1000000; 0 = tous les chemins) ; doit être non négatif
• decay_factor (float) : décroissance du classement par étape ; 2 signifie la réduction de moitié (valeur par défaut : 1) ; doit être non négative
• is_directional (bool) : indique si les arêtes sont directionnelles (valeur par défaut : True)
• min_hop_count (int) : tronçons minimaux (valeur par défaut : 1) ; doivent être ≥ 1
• max_hop_count (int) : le nombre maximal de sauts (valeur par défaut : 4) doit être ≥ min_hop_count
• shortest_path (bool) : retourne uniquement les chemins d’accès les plus courts (par défaut : False)
• max_results (int) : le nombre maximal de résultats (par défaut : 500) doit être ≥ 1

Soulève:

• ValueError: si rank_property_name est manquant, threshold < 0, max_paths < 0, decay_factor < 0min_hop_count < 1, max_hop_count < min_hop_count, oumax_results < 1
• RuntimeError: si l’initialisation du client ou l’exécution de requête échoue

Retourne:

• QueryResult: contenant les nœuds/arêtes classés

Exemple :

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

Convertissez l’intégralité du graphique en GraphFrame. Utilise les données de spécification lorsqu’elles sont disponibles ; lit à partir des tables de lac dans le cas contraire.

Paramètres :

• column_mapping (Dict[str, str], facultatif) : mappage de colonnes personnalisé

Retourne:

• GraphFrame: objet GraphFrame avec tous les sommets et bords

Exemple :

gf = graph.to_graphframe()

show

def show() -> None

Afficher les informations du graphique. Délègue à spec.show() pour un affichage enrichi lorsqu’une spécification est attachée ; dans le cas contraire, affiche des informations minimales.

show_schema

def show_schema() -> None

Afficher le schéma de graphe. Délègue à spec.show_schema() lorsqu’une spécification est jointe ; imprime un message indiquant qu’aucun schéma n’est disponible dans le cas contraire.

publish (nouveauté de la version 0.3.3)

def publish() -> Graph

Inscrivez le graphe auprès de l’API, ce qui le rend interrogeable. Appelez ceci après Graph.prepare() (ou sur tout Graph objet associé à une spécification) pour publier le graphique instance.

Retourne:

• Graph: Self pour le chaînage de méthodes

Soulève:

• ValueError: si aucune spécification n’est attachée ou si le contexte est manquant
• RuntimeError: si la publication échoue

Exemple :

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

BuildStatus

Dataclass qui transporte les métadonnées d’une Graph.build() opération.

Champs

Chemins de construction

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)

Exemple :

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

Générateurs de nœuds

NodeBuilderInitial

État initial pour le générateur de nœuds : seules les méthodes de source de données sont disponibles.

Constructeur

NodeBuilderInitial(alias: str, graph_builder: GraphSpecBuilder)

Note: Généralement créé via GraphSpecBuilder.add_node(), pas directement instancié.

Méthodes

from_table

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

Définissez la table comme source de données avec une résolution de base de données intelligente.

Paramètres :

• table_name (str) : nom de la table (obligatoire)
• database (str, facultatif) : nom de base de données explicite (prioritaire sur le contexte par défaut)

Retourne:

• NodeBuilderSourceSet: Générateur pour une configuration supplémentaire

Soulève:

• ValueError: si la table est introuvable ou si plusieurs tables en conflit ont été trouvées

Ordre de résolution de base de données :

1. Paramètre explicite database (priorité la plus élevée)
2. ExecutionContext.default_database
3. Rechercher dans toutes les bases de données (avec détection de conflit)

Exemple :

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

from_dataframe

def from_dataframe(dataframe: DataFrame) -> NodeBuilderSourceSet

Définissez Le DataFrame Spark comme source de données.

Paramètres :

• dataframe (DataFrame) : DataFrame Spark

Retourne:

• NodeBuilderSourceSet: Générateur pour une configuration supplémentaire

Exemple :

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

NodeBuilderSourceSet

Générateur de nœuds après la définition de la source de données : méthodes de configuration disponibles.

Constructeur

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

Note: Créé en interne par les méthodes source NodeBuilderInitial.

Méthodes

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

Appliquez le filtrage d’intervalle de temps à la source de données du nœud.

Paramètres :

• time_column (str) : nom de colonne contenant des données d’horodatage (obligatoire)
• start_time (str ou datetime, facultatif) : date de début ('20/10/25', '2025-10-20', ou objet datetime)
• end_time (str ou datetime, facultatif) : date de fin (mêmes formats que start_time)
• lookback_hours (float, facultatif) : heures pour revenir en arrière à partir de maintenant

Retourne:

• NodeBuilderSourceSet: Self pour le chaînage de méthodes

Soulève:

• ValueError: si la colonne de temps est introuvable dans le schéma source

Logique d’intervalle de temps :

1. Si start_time et end_time fournis : utilisez-les directement
2. Si seul lookback_hours fourni : end=now, start=now-lookback_hours
3. Si rien n’est fourni : aucun filtrage de temps
4. Si start/end AND lookback_hours : start/end sont prioritaires

Exemple :

# 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

Définissez l’étiquette du nœud (la valeur par défaut est alias si elle n’est pas appelée).

Paramètres :

• label (str) : étiquette de nœud

Retourne:

• NodeBuilderSourceSet: Self pour le chaînage de méthodes

Soulève:

• ValueError: si l’étiquette est déjà définie

Exemple :

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

with_columns

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

Configurez les colonnes avec la clé et la désignation d’affichage requises.

Paramètres :

• *columns (str) : noms de colonnes à inclure (au moins un obligatoire)
• key (str) : nom de colonne à marquer comme clé (obligatoire, doit être dans les colonnes)
• display (str) : nom de colonne à marquer comme valeur d’affichage (obligatoire, doit être en colonnes, peut être identique à la clé)

Retourne:

• NodeBuilderSourceSet: Self pour le chaînage de méthodes

Soulève:

• ValueError: échec de la validation (colonnes dupliquées, clé/affichage manquant, etc.)

Remarques :

• Les propriétés sont générées automatiquement à partir de types de colonnes

• La colonne de filtre de temps est automatiquement ajoutée si elle est spécifiée

• Les types de propriétés sont déduits automatiquement à partir du schéma source

• Voir Restrictions

Exemple :

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

Terminez ce nœud et commencez à créer un autre nœud.

Paramètres :

• alias (str) : alias du nouveau nœud

Retourne:

• NodeBuilderInitial: Nouveau générateur de nœuds

Exemple :

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

Terminez ce nœud et commencez à créer une arête.

Paramètres :

• alias (str) : alias de l’arête

Retourne:

• EdgeBuilderInitial: Nouveau générateur d’arêtes

Exemple :

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

done

def done() -> GraphSpec

Finalisez ce nœud et complétez la spécification du graphique.

Retourne:

• GraphSpec: spécification complète du graphique

Exemple :

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

Générateurs Edge

EdgeBuilderInitial

État initial pour edge Builder : seules les méthodes de source de données disponibles.

Constructeur

EdgeBuilderInitial(alias: str, graph_builder: GraphSpecBuilder)

Note: Généralement créé via GraphSpecBuilder.add_edge(), pas directement instancié.

Méthodes

from_table

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

Définissez la table comme source de données avec une résolution de base de données intelligente.

Paramètres :

• table_name (str) : nom de la table (obligatoire)
• database (str, facultatif) : nom de base de données explicite

Retourne:

• EdgeBuilderSourceSet: Générateur pour une configuration supplémentaire

Soulève:

• ValueError: si la table est introuvable ou si plusieurs tables en conflit ont été trouvées

Exemple :

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

from_dataframe

def from_dataframe(dataframe: DataFrame) -> EdgeBuilderSourceSet

Définissez Le DataFrame Spark comme source de données.

Paramètres :

• dataframe (DataFrame) : DataFrame Spark

Retourne:

• EdgeBuilderSourceSet: Générateur pour une configuration supplémentaire

Exemple :

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

EdgeBuilderSourceSet

Générateur Edge une fois la source de données définie : méthodes de configuration disponibles.

Constructeur

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

Note: Créé en interne par les méthodes sources EdgeBuilderInitial.

Méthodes

with_label

def with_label(label: str) -> EdgeBuilderSourceSet

Définissez le type/l’étiquette de relation d’arête (alias par défaut s’il n’est pas appelé).

Paramètres :

• label (str) : étiquette d’arête

Retourne:

• EdgeBuilderSourceSet: Self pour le chaînage de méthodes

Soulève:

• ValueError: si l’étiquette est déjà définie

Exemple :

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

edge_label

def edge_label(label: str) -> EdgeBuilderSourceSet

Définissez le type/l’étiquette de relation d’arête (alias par défaut s’il n’est pas appelé).

Paramètres :

• label (str) : étiquette d’arête

Retourne:

• EdgeBuilderSourceSet: Self pour le chaînage de méthodes

Soulève:

• ValueError: si l’étiquette est déjà définie

Exemple :

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

source

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

Définissez le nœud source avec la colonne d’ID et l’étiquette.

Paramètres :

• id_column (str) : nom de colonne contenant l’ID de nœud source
• node_type (str) : étiquette du nœud source

Retourne:

• EdgeBuilderSourceSet: Self pour le chaînage de méthodes

Soulève:

• ValueError: si la source est déjà définie

Exemple :

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

Définissez le nœud cible avec la colonne d’ID et l’étiquette.

Paramètres :

• id_column (str) : nom de colonne contenant l’ID du nœud cible
• node_type (str) : étiquette de nœud cible

Retourne:

• EdgeBuilderSourceSet: Self pour le chaînage de méthodes

Soulève:

• ValueError: si la cible est déjà définie

Exemple :

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

Appliquez le filtrage d’intervalle de temps à la source de données de la périphérie.

Paramètres :

• time_column (str) : nom de colonne contenant des données d’horodatage (obligatoire)
• start_time (str ou datetime, facultatif) : date de début
• end_time (str ou datetime, facultatif) : Date de fin
• lookback_hours (float, facultatif) : heures pour revenir en arrière à partir de maintenant

Retourne:

• EdgeBuilderSourceSet: Self pour le chaînage de méthodes

Soulève:

• ValueError: si la colonne de temps est introuvable dans le schéma source

Exemple :

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

Configurez les colonnes avec la clé et la désignation d’affichage requises.

Paramètres :

• *columns (str) : noms de colonnes à inclure (au moins un obligatoire)
• key (str) : nom de colonne à marquer comme clé (obligatoire, doit être dans les colonnes)
• display (str) : nom de colonne à marquer comme valeur d’affichage (obligatoire, doit être en colonnes)

Retourne:

• EdgeBuilderSourceSet: Self pour le chaînage de méthodes

Soulève:

• ValueError: si la validation échoue

Exemple :

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

Terminez ce bord et commencez à créer un nœud.

Paramètres :

• alias (str) : alias du nouveau nœud

Retourne:

• NodeBuilderInitial: Nouveau générateur de nœuds

add_edge

def add_edge(alias: str) -> EdgeBuilderInitial

Terminez ce bord et commencez à créer un autre bord.

Paramètres :

• alias (str) : alias du nouveau bord

Retourne:

• EdgeBuilderInitial: Nouveau générateur d’arêtes

Exemple :

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

Finalisez cette arête et complétez la spécification du graphique.

Retourne:

• GraphSpec: spécification complète du graphique

Classes de schéma

GraphDefinitionReference

Référence à une définition de graphique avec le nom et la version.

Constructeur

GraphDefinitionReference(
    fully_qualified_name: str,
    version: str
)

Paramètres :

• fully_qualified_name (str) : nom complet du graphe référencé
• version (str) : version du graphe référencé

Soulève:

• ValueError: si fully_qualified_name ou la version est vide

Méthodes

to_dict

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

Sérialiser dans le dictionnaire.

Retourne:

• Dict[str, Any]: référence sérialisée

Propriété

Définition de propriété avec interface de type sécurisé.

Constructeur

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
)

Paramètres :

• name (str) : nom de la propriété
• property_type (PropertyType) : type de données Property
• is_non_null (bool, default=False) : indique si la propriété est requise
• description (str, default=" ») : Description de la propriété
• is_key (bool, default=False) : indique si la propriété est une clé
• is_display_value (bool, default=False) : indique si la propriété est une valeur d’affichage
• is_internal (bool, default=False) : indique si la propriété est interne

Soulève:

• ValueError: si name est vide ou si la validation échoue

Méthodes de classe

key

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

Créez une propriété de clé avec des paramètres communs (is_key=True, is_display_value=True).

display

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

Créez une propriété de valeur d’affichage (is_display_value=True).

Méthodes

describe

def describe(text: str) -> Property

Ajoutez une description couramment.

Paramètres :

• text (str) : texte de description

Retourne:

• Property: Self pour le chaînage de méthodes

to_dict

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

Sérialisez la propriété dans le dictionnaire avec des clés d’annotation @-préfixe.

Retourne:

• Dict[str, Any]: propriété sérialisée

to_gql

def to_gql() -> str

Générer une définition de propriété GQL.

Retourne:

• str: représentation sous forme de chaîne GQL

EdgeNode

Référence de nœud utilisée dans les définitions d’arête.

Constructeur

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

Paramètres :

• alias (str, facultatif) : alias de nœud (défini automatiquement sur la première étiquette si Aucun ou vide)
• labels (List[str]) : étiquettes de nœud (au moins une obligatoire)

Soulève:

• ValueError: si la liste des étiquettes est vide
• TypeError: si les étiquettes ne sont pas des chaînes

Auto-mutation :

• Si alias a la valeur None ou est vide, il est défini sur la première étiquette

Méthodes

to_dict

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

Sérialiser dans le dictionnaire.

Retourne:

• Dict[str, Any]: Référence de nœud de périphérie sérialisé

Nœud

Définition de nœud avec interface de type sécurisé.

Constructeur

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
)

Paramètres :

• alias (str, default=" ») : alias de nœud (défini automatiquement sur la première étiquette si vide)
• labels (List[str]) : étiquettes de nœud (au moins une obligatoire)
• implies_labels (List[str], default=[]) : étiquettes implicites
• properties (List[Property], default=[]) : propriétés du nœud
• description (str, default=" ») : Description du nœud
• entity_group (str, default=" ») : nom du groupe d’entités
• dynamic_labels (bool, default=False) : indique si le nœud a des étiquettes dynamiques
• abstract_edge_aliases (bool, default=False) : indique si le nœud utilise des alias de périphérie abstraits

Soulève:

• ValueError: si la validation échoue (aucune étiquette, aucune propriété de clé, aucune propriété d’affichage, etc.)

Auto-mutation :

• Si alias est vide, il est défini sur la première étiquette
• Si entity_group est vide, elle est définie sur l’étiquette principale

Méthodes

get_primary_label

def get_primary_label() -> Optional[str]

Obtenez l’étiquette principale (première).

Retourne:

• str ou None: étiquette principale ou Aucun si aucune étiquette

get_entity_group_name

def get_entity_group_name() -> str

Obtenez le nom du groupe d’entités ou le secours vers l’étiquette principale.

Retourne:

• str: nom du groupe d’entités

get_primary_key_property_name

def get_primary_key_property_name() -> Optional[str]

Obtient le nom de la propriété de clé primaire.

Retourne:

• str ou None: nom de la propriété de clé primaire

get_properties

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

Obtenez les propriétés sous forme de dictionnaire pour un accès facile.

Retourne:

• Dict[str, Property]: propriétés clé par nom

get_property

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

Obtient une propriété spécifique par nom.

Paramètres :

• name (str) : nom de la propriété

Retourne:

• Property ou None: propriété si elle est trouvée

add_property

def add_property(prop: Property) -> None

Ajoutez une propriété à ce nœud.

Paramètres :

• prop (Propriété) : propriété à ajouter

Soulève:

• ValueError: si le nom de la propriété est dupliqué

is_dynamically_labeled

def is_dynamically_labeled() -> bool

Vérifiez si le nœud a des étiquettes dynamiques.

Retourne:

• bool: True si les étiquettes dynamiques sont activées

is_abstract_edge_node_aliases

def is_abstract_edge_node_aliases() -> bool

Vérifiez si le nœud utilise des alias de nœud de périphérie abstraits.

Retourne:

• bool: True si les alias de périmètre abstraits sont activés

describe

def describe(text: str) -> Node

Ajoutez une description couramment.

Paramètres :

• text (str) : texte de description

Retourne:

• Node: Self pour le chaînage de méthodes

to_dict

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

Sérialiser le nœud dans le dictionnaire.

Retourne:

• Dict[str, Any]: nœud sérialisé

to_gql

def to_gql() -> str

Générer une définition de nœud GQL.

Retourne:

• str: représentation sous forme de chaîne GQL

Soulève:

• ValueError: si le nœud n’a pas de champs requis pour GQL

Méthodes de classe

create

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

Créez un nœud avec tous les champs obligatoires.

Paramètres :

• alias (str) : alias de nœud
• labels (List[str]) : étiquettes de nœud
• properties (List[Property]) : propriétés du nœud
• description (str, default=" ») : Description du nœud
• entity_group (str, default=" ») : nom du groupe d’entités

Retourne:

• Node: nouveau nœud instance

Microsoft Edge

Définition d’arête avec interface de type sécurisé.

Constructeur

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
)

Paramètres :

• relationship_type (str) : type de relation Edge (par exemple, « FOLLOWS », « OWNS »)
• source_node_label (str) : étiquette du nœud source
• target_node_label (str) : étiquette de nœud cible
• direction (EdgeDirection, default=DIRECTED_RIGHT) : direction de l’arête
• properties (List[Property], default=[]) : propriétés Edge
• description (str, default=" ») : Description de l’arête
• entity_group (str, default=" ») : nom du groupe d’entités
• dynamic_type (bool, default=False) : indique si edge a un type dynamique

Soulève:

• ValueError: si la validation échoue

Auto-mutation :

• labels la liste est automatiquement remplie avec [relationship_type]
• Si entity_group est vide, il est défini sur relationship_type

Propriétés

edge_type

def edge_type() -> str

Alias de compatibilité descendante pour relationship_type.

Retourne:

• str: Type de relation

Méthodes

get_entity_group_name

def get_entity_group_name() -> str

Obtenez le nom du groupe d’entités ou le type de relation de secours.

Retourne:

• str: nom du groupe d’entités

is_dynamic_type

def is_dynamic_type() -> bool

Vérifiez si edge a un type dynamique.

Retourne:

• bool: True si le type dynamique

add_property

def add_property(edge_property: Property) -> None

Ajoutez une propriété à ce bord.

Paramètres :

• edge_property (Propriété) : propriété à ajouter

describe

def describe(text: str) -> Edge

Ajoutez une description couramment.

Paramètres :

• text (str) : texte de description

Retourne:

• Edge: Self pour le chaînage de méthodes

to_dict

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

Sérialiser edge en dictionnaire.

Retourne:

• Dict[str, Any]: arête sérialisée

to_gql

def to_gql() -> str

Générer une définition de bord GQL.

Retourne:

• str: représentation sous forme de chaîne GQL

Méthodes de classe

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

Créez une arête avec tous les champs obligatoires.

Paramètres :

• relationship_type (str) : type de relation Edge
• source_node_label (str) : étiquette du nœud source
• target_node_label (str) : étiquette de nœud cible
• properties (List[Property], facultatif) : propriétés Edge
• description (str, default=" ») : Description de l’arête
• entity_group (str, default=" ») : nom du groupe d’entités

Retourne:

• Edge: Nouvelle instance edge

GraphSchema

Définition de schéma de graphe avec interface de type sécurisé.

Constructeur

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

Paramètres :

• name (str) : nom du schéma de graphique
• nodes (List[Node], default=[]) : définitions de nœud
• edges (List[Edge], default=[]) : définitions edge
• base_graphs (List[GraphSchema], default=[]) : schémas de graphe de base
• description (str, default=" ») : Description du schéma
• version (str, default="1.0 ») : version du schéma
• fully_qualified_name (str, default=" ») : nom complet
• namespace (str, default=" ») : espace de noms

Soulève:

• ValueError: si la validation échoue (alias dupliqués, arêtes référencent des nœuds inexistants, etc.)

Méthodes

get_fully_qualified_name

def get_fully_qualified_name() -> str

Obtenez un nom complet.

Retourne:

• str: nom complet

get_namespace

def get_namespace() -> str

Obtenez l’espace de noms à partir du nom complet ou retournez la valeur par défaut.

Retourne:

• str:Noms

get_version

def get_version() -> str

Obtenir la version.

Retourne:

• str: Chaîne de version

get_node

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

Obtenir le nœud par étiquette ou alias.

Paramètres :

• label_or_alias (str) : étiquette ou alias de nœud

Retourne:

• Node ou None: nœud s’il est trouvé

get_edge

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

Obtenir l’arête par nom/type.

Paramètres :

• name (str) : type de relation Edge

Retourne:

• Edge ou None: Edge s’il est trouvé

add_node

def add_node(node: Node) -> None

Ajoutez un nœud à ce graphique.

Paramètres :

• node (Nœud) : nœud à ajouter

Soulève:

• ValueError: si l’alias de nœud est dupliqué

add_edge

def add_edge(edge: Edge) -> None

Ajoutez un bord à ce graphique.

Paramètres :

• edge (Edge) : Edge à ajouter

Soulève:

• ValueError: si le type d’arête est dupliqué

include_graph

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

Ajoutez un graphe include (API Fluent).

Paramètres :

• fully_qualified_name (str) : nom complet du graphe à inclure
• version (str) : version du graphique à inclure

Retourne:

• GraphSchema: Self pour le chaînage de méthodes

get_included_graph_references

def get_included_graph_references() -> List[GraphDefinitionReference]

Obtenez la liste des références de graphe incluses.

Retourne:

• List[GraphDefinitionReference]: Liste des références de définition de graphique

describe

def describe(text: str) -> GraphSchema

Ajoutez une description couramment.

Paramètres :

• text (str) : texte de description

Retourne:

• GraphSchema: Self pour le chaînage de méthodes

to_dict

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

Sérialiser le schéma dans le dictionnaire.

Retourne:

• Dict[str, Any]: schéma sérialisé

to_json

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

Générer une représentation JSON.

Paramètres :

• indent (int, default=2) : niveau de mise en retrait JSON

Retourne:

• str: chaîne JSON

to_gql

def to_gql() -> str

Générez une définition de schéma GQL.

Retourne:

• str: représentation sous forme de chaîne GQL

Méthodes de classe

create

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

Créez un schéma de graphe avec tous les champs obligatoires.

Paramètres :

• name (str) : nom du schéma de graphique
• nodes (List[Node], facultatif) : Définitions de nœud
• edges (Liste[Edge], facultatif) : définitions edge
• description (str, default=" ») : Description du schéma
• version (str, default="1.0 ») : version du schéma

Retourne:

• GraphSchema: nouvelle instance de schéma de graphe

Classes d’entrée de requête

Classes de données représentant les paramètres d’entrée pour les requêtes de graphe prédéfinies.

QueryInputBase

Classe de base pour tous les paramètres d’entrée de requête.

Méthodes

to_json_payload

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

Convertissez les paramètres d’entrée en dictionnaire pour l’envoi d’API.

Retourne:

• Dict[str, Any]: représentation par dictionnaire des paramètres d’entrée

validate

def validate() -> None

Validez les paramètres d’entrée.

Soulève:

• ValueError: si les paramètres d’entrée ne sont pas valides

ReachabilityQueryInput

Paramètres d’entrée pour une requête d’accessibilité entre les nœuds source et cible. Hérite de ReachabilityQueryInputBase qui hérite de QueryInputBase.

Champs

Validation:

• source_property_value est obligatoire
• target_property_value est obligatoire

Exemple :

# 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

Paramètres d’entrée d’une requête k-hop à partir d’un nœud source donné. Hérite de ReachabilityQueryInputBase.

Hérite de tous les champs de ReachabilityQueryInput.

Validation:

• Au moins un de source_property_value ou target_property_value doit être fourni

Exemple :

# 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

Paramètres d’entrée d’une requête de rayon d’explosion de la source vers les nœuds cibles. Hérite de ReachabilityQueryInputBase.

Hérite de tous les champs de ReachabilityQueryInput, avec les champs obligatoires suivants :

Validation:

• source_property_value est obligatoire
• target_property_value est obligatoire

Exemple :

# 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

Paramètres d’entrée pour une requête d’analyse de centralité. Hérite de QueryInputBase.

CentralityType Enum

Champs

Exemple :

# 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

Paramètres d’entrée pour une requête d’analyse classée. Hérite de QueryInputBase.

Champs

Exemple :

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

Résultats de la requête

QueryResult

Résultat d’une requête de graphe avec un accès différé au DataFrame.

Constructeur

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

Paramètres :

• raw_response (Dict[str, Any]) : dictionnaire de réponse d’API brut
• graph (Graph) : référence au graphe parent

Note: Généralement créé par Graph.query(), pas directement instancié.

Méthodes

to_dataframe

def to_dataframe() -> DataFrame

Convertit le résultat de la requête en dataframe Spark.

Retourne:

• DataFrame: Résultat de la requête en tant que DataFrame Spark

Soulève:

• ValueError: si la conversion échoue

Exemple :

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

get_raw_data

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

Obtenir la section RawData à partir de la réponse.

Retourne:

• Dict[str, Any]: dictionnaire avec métadonnées brutes ou dict vide s’il n’est pas présent

Exemple :

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

show

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

Afficher le résultat de la requête dans différents formats.

Paramètres :

• format (str, default="visual ») : Format de sortie
  • "table": tables de DataFrame complètes (toutes les colonnes)
  • "visual": visualisation graphique interactive avec plug-in VSC
  • "all": Afficher tous les formats

Soulève:

• ValueError: si format n’est pas l’une des valeurs prises en charge

Exemple :

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

Exemple complet (recommandé — 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()

Remarques sur les modèles de conception

Fluent API

Tous les générateurs prennent en charge le chaînage de méthodes pour les définitions de graphe déclaratives et lisibles :

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

Schémas d’union

Plusieurs arêtes avec le même alias sont automatiquement unionées avec les propriétés fusionnées :

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

Configuration automatique

De nombreux champs ont des valeurs par défaut raisonnables :

• Les étiquettes nœud/arête sont par défaut leurs alias
• Les propriétés sont automatiquement déduites à partir des schémas sources
• Les groupes d’entités par défaut sont les étiquettes/types de relation principaux

Évaluation différée

Les dataframes et les ressources sont chargés de manière différée et mis en cache :

• graph_spec.nodes et graph_spec.edges sont chargés lors du premier accès
• Les résultats de la requête créent des DataFrames uniquement à la demande

Contenu connexe

• Vue d’ensemble des graphiques personnalisés
• Créer des graphiques personnalisés