#!/usr/bin/env python
from __future__ import annotations
from typing import Iterable
from sentry.backup.scopes import RelocationScope
from sentry.runner import configure
from sentry.silo.base import SiloMode
configure()
from enum import Enum, unique
from string import Template
import click
from django.db import models
from sentry.backup.dependencies import ForeignField, ForeignFieldKind, ModelRelations, dependencies
digraph = Template(
"""
digraph Models {
ranksep = 8;
rankdir=LR
node [style="rounded,filled",shape="rectangle"];
subgraph cluster_legend {
label = "Legend";
fontsize="40"
node [shape="plaintext",style="none"]
key1 [label=<
HybridCloudForeignKey |
Explicit ForeignKey |
Implicit ForeignKey |
Control Silo Model |
Region Silo Model |
Unexported Model |
>]
key2 [label=<>]
key1:i1:e -> key2:i1:w [color=green]
key1:i2:e -> key2:i2:w [color=blue]
key1:i3:e -> key2:i3:w [color=red]
}
$clusters
$edges
}
"""
)
cluster = Template(
"""
subgraph cluster_$num {
label="$name Relocation Scope"
style="rounded,filled"
shape="rectangle"
fillcolor="$fill"
fontsize="40"
color="grey"
$nodes
}
"""
)
@unique
class ClusterColor(Enum):
Purple = "lavenderblush"
Yellow = "khaki"
Green = "honeydew"
Blue = "lightsteelblue1"
@unique
class NodeColor(Enum):
Red = "lightpink"
Blue = "lightblue"
@unique
class EdgeColor(Enum):
Hybrid = "color=green"
Explicit = "color=blue"
Implicit = "color=red"
def print_model_node(model: models.base.ModelBase, silo: SiloMode) -> str:
color = NodeColor.Red if silo == SiloMode.CONTROL else NodeColor.Blue
return f""""{model.__name__}" [fillcolor="{color.value}"];"""
def print_rel_scope_subgraph(
name: str, num: int, rels: Iterable[ModelRelations], color: ClusterColor
) -> str:
return cluster.substitute(
num=num,
name=name,
fill=color.value,
nodes="\n ".join([print_model_node(mr.model, mr.silos[0]) for mr in rels]),
)
def print_edges(mr: ModelRelations) -> str:
if len(mr.foreign_keys) == 0:
return ""
src = mr.model
return "\n ".join([print_edge(src, ff.model, ff) for ff in mr.foreign_keys.values()])
def print_edge(src: models.base.ModelBase, dest: models.base.ModelBase, field: ForeignField) -> str:
color = EdgeColor.Explicit
if field.kind == ForeignFieldKind.HybridCloudForeignKey:
color = EdgeColor.Hybrid
elif field.kind == ForeignFieldKind.ImplicitForeignKey:
color = EdgeColor.Implicit
style = "dashed" if field.nullable else "solid"
return f""""{src.__name__}":e -> "{dest.__name__}":w [{color.value},style={style}];"""
def get_most_permissive_relocation_scope(mr: ModelRelations) -> RelocationScope:
if isinstance(mr.relocation_scope, set):
return sorted(list(mr.relocation_scope), key=lambda obj: obj.value * -1)[0]
return mr.relocation_scope
@click.command()
@click.option("--show-excluded", default=False, is_flag=True, help="Show unexportable models too")
def main(show_excluded: bool):
"""Generate a graphviz spec for the current model dependency graph."""
# Get all dependencies, filtering as necessary.
deps = sorted(dependencies().values(), key=lambda mr: mr.model.__name__)
if not show_excluded:
deps = list(filter(lambda m: m.relocation_scope != RelocationScope.Excluded, deps))
# Group by most permissive region scope.
user_scoped = filter(
lambda m: get_most_permissive_relocation_scope(m) == RelocationScope.User, deps
)
org_scoped = filter(
lambda m: get_most_permissive_relocation_scope(m) == RelocationScope.Organization, deps
)
config_scoped = filter(
lambda m: get_most_permissive_relocation_scope(m) == RelocationScope.Config, deps
)
global_scoped = filter(
lambda m: get_most_permissive_relocation_scope(m) == RelocationScope.Global, deps
)
# Print nodes.
clusters = "".join(
[
print_rel_scope_subgraph("User", 1, user_scoped, ClusterColor.Green),
print_rel_scope_subgraph("Organization", 2, org_scoped, ClusterColor.Purple),
print_rel_scope_subgraph("Config", 3, config_scoped, ClusterColor.Blue),
print_rel_scope_subgraph("Global", 4, global_scoped, ClusterColor.Yellow),
]
)
# Print edges.
edges = "\n ".join(filter(lambda s: s, [print_edges(mr) for mr in deps]))
click.echo(digraph.substitute(clusters=clusters, edges=edges))
if __name__ == "__main__":
main()