Skip to content

Commit

Permalink
chore: better separation of example usage in nextgen workspace
Browse files Browse the repository at this point in the history
  • Loading branch information
z3z1ma committed Jul 20, 2024
1 parent 0017529 commit 579fc9e
Showing 1 changed file with 50 additions and 48 deletions.
98 changes: 50 additions & 48 deletions src/cdf/nextgen/workspace.py
Original file line number Diff line number Diff line change
Expand Up @@ -87,51 +87,53 @@ def add_dependency(self, name: DependencyKey, definition: Dependency) -> None:
self.injector.add_definition(name, definition)


class DataTeamWorkspace(Workspace):
name = "data-team"
version = "0.1.1"

def get_services(self):
return {
"a": Dependency(1),
"b": Dependency(lambda a: a + 1),
"prod_bigquery": Dependency("dwh-123"),
"sfdc": Dependency(
ConfigResolver.map_section("sfdc")(
lambda username: f"https://sfdc.com/{username}"
)
),
}

def get_config_sources(self):
return [
{
"sfdc": {"username": "abc"},
"bigquery": {"project_id": ...},
},
*super().get_config_sources(),
]


datateam = DataTeamWorkspace()


@ConfigResolver.map_values(b="a.b.c")
def c(b: int) -> int:
return b * 10


datateam.add_dependency("c", Dependency(c))


def source_a(a: int, prod_bigquery: str):
print(f"Source A: {a=}, {prod_bigquery=}")


print(datateam.injector(source_a))
print(datateam.name)
print(datateam.config_resolver["sfdc.username"])

print(datateam.injector.get_or_raise("sfdc"))

datateam.cli()
if __name__ == "__main__":
# Example usage

# Define a workspace
class DataTeamWorkspace(Workspace):
name = "data-team"
version = "0.1.1"

def get_services(self):
return {
"a": Dependency(1),
"b": Dependency(lambda a: a + 1),
"prod_bigquery": Dependency("dwh-123"),
"sfdc": Dependency(
ConfigResolver.map_section("sfdc")(
lambda username: f"https://sfdc.com/{username}"
)
),
}

def get_config_sources(self):
return [
{
"sfdc": {"username": "abc"},
"bigquery": {"project_id": ...},
},
*super().get_config_sources(),
]

# Create an instance of the workspace
datateam = DataTeamWorkspace()

@ConfigResolver.map_values(b="a.b.c")
def c(b: int) -> int:
return b * 10

# Imperatively add dependencies
datateam.add_dependency("c", Dependency(c))

def source_a(a: int, prod_bigquery: str):
print(f"Source A: {a=}, {prod_bigquery=}")

# Some interface examples
print(datateam.injector(source_a))
print(datateam.name)
print(datateam.config_resolver["sfdc.username"])
print(datateam.injector.get_or_raise("sfdc"))

# Run the autogenerated CLI
datateam.cli()

0 comments on commit 579fc9e

Please sign in to comment.