lafrite/plesna
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: build dataplatform graph and graphset dynamicaly

Browse Source
This commit is contained in:
Bertrand Benjamin 2025-01-05 16:25:22 +01:00
parent f2ed76c8aa
commit 44a7eed5b4
1 changed files with 18 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
plesna/dataplatform.py
View File

@ -12,31 +12,15 @@ class DataPlateformError(Exception):
class DataPlateform: class DataPlateform:
def __init__(self): def __init__(self):
self._graphset = GraphSet()
self._graph = Graph()
self._metadata_engine = "" self._metadata_engine = ""
self._fluxes = {} self._fluxes = {}
self._repositories = {} self._repositories = {}
@property
def graphset(self) -> GraphSet:
return self._graphset
@property
def graph(self) -> Graph:
return self._graph
def repository_graph_feed(self, repository_id: str):
for schema in self._repositories[repository_id].schemas():
for table in self._repositories[repository_id].tables(schema):
self._graph.add_node(Node(name=table))
def add_repository(self, repository: Repository) -> str: def add_repository(self, repository: Repository) -> str:
if repository.id in self._repositories: if repository.id in self._repositories:
raise DataPlateformError("The repository {repository.id} already exists") raise DataPlateformError("The repository {repository.id} already exists")
self._repositories[repository.id] = repository self._repositories[repository.id] = repository
self.repository_graph_feed(repository.id)
return repository.id return repository.id
@property @property
@ -51,14 +35,8 @@ class DataPlateform:
raise DataPlateformError("The flux {name} already exists") raise DataPlateformError("The flux {name} already exists")
self._fluxes[name] = flux self._fluxes[name] = flux
self.flux_graphset_feed(name)
return name return name
def flux_graphset_feed(self, flux_name: str):
flux = self.flux(flux_name)
edge = flux_to_edgeonset(flux)
self._graphset.append(edge)
@property @property
def fluxes(self) -> list[str]: def fluxes(self) -> list[str]:
return list(self._fluxes) return list(self._fluxes)
@ -70,3 +48,21 @@ class DataPlateform:
if name not in self._fluxes: if name not in self._fluxes:
raise DataPlateformError("The flux {name} is not registered") raise DataPlateformError("The flux {name} is not registered")
return consume_flux(self._fluxes[name]) return consume_flux(self._fluxes[name])
@property
def graphset(self) -> GraphSet:
graphset = GraphSet()
for flux in self._fluxes.values():
edge = flux_to_edgeonset(flux)
graphset.append(edge)
return graphset
@property
def graph(self) -> Graph:
graph = Graph()
for repo in self._repositories.values():
for schema in repo.schemas():
for table in repo.tables(schema):
graph.add_node(Node(name=table))
return graph