feat: add DI following a singleton pattern

Author: GermanMT

src/dagster_app/__init__.py

@@ -1,39 +1,11 @@
 from dagster import Definitions, load_assets_from_modules
 
 from src.dagster_app import assets
-from src.dagster_app.resources import (
-    AttributorResource,
-    CargoServiceResource,
-    MavenServiceResource,
-    NPMServiceResource,
-    NuGetServiceResource,
-    PackageServiceResource,
-    PyPIServiceResource,
-    RubyGemsServiceResource,
-    VersionServiceResource,
-    VulnerabilityServiceResource,
-)
 from src.dagster_app.schedules import all_schedules
 
 all_assets = load_assets_from_modules([assets])
 
-vuln_service = VulnerabilityServiceResource()
-
-resources = {
-    "pypi_service": PyPIServiceResource(),
-    "npm_service": NPMServiceResource(),
-    "maven_service": MavenServiceResource(),
-    "cargo_service": CargoServiceResource(),
-    "rubygems_service": RubyGemsServiceResource(),
-    "nuget_service": NuGetServiceResource(),
-    "package_service": PackageServiceResource(),
-    "version_service": VersionServiceResource(),
-    "vulnerability_service": vuln_service,
-    "attributor": AttributorResource(vuln_service=vuln_service),
-}
-
 defs = Definitions(
     assets=all_assets,
     schedules=all_schedules,
-    resources=resources,
 )

src/dagster_app/assets/cargo_assets.py

@@ -3,11 +3,11 @@
 
 from dagster import AssetExecutionContext, MetadataValue, Output, asset
 
-from src.dagster_app.resources import (
-    AttributorResource,
-    CargoServiceResource,
-    PackageServiceResource,
-    VersionServiceResource,
+from src.dependencies import (
+    get_attributor,
+    get_cargo_service,
+    get_package_service,
+    get_version_service,
 )
 from src.logger import logger
 from src.processes.extractors import CargoPackageExtractor
@@ -22,18 +22,14 @@
 )
 def cargo_package_ingestion(
     context: AssetExecutionContext,
-    cargo_service: CargoServiceResource,
-    package_service: PackageServiceResource,
-    version_service: VersionServiceResource,
-    attributor: AttributorResource,
 ) -> Output[dict[str, Any]]:
     try:
         logger.info("Starting Cargo package ingestion process")
 
-        cargo_svc = cargo_service.get_service()
-        package_svc = package_service.get_service()
-        version_svc = version_service.get_service()
-        attr = attributor.get_attributor()
+        cargo_svc = get_cargo_service()
+        package_svc = get_package_service()
+        version_svc = get_version_service()
+        attr = get_attributor()
 
         async def _run():
             new_packages = 0
@@ -118,18 +114,14 @@ async def _run():
 )
 def cargo_packages_updates(
     context: AssetExecutionContext,
-    cargo_service: CargoServiceResource,
-    package_service: PackageServiceResource,
-    version_service: VersionServiceResource,
-    attributor: AttributorResource,
 ) -> Output[dict[str, Any]]:
     try:
         logger.info("Starting Cargo package version update process")
 
-        cargo_svc = cargo_service.get_service()
-        package_svc = package_service.get_service()
-        version_svc = version_service.get_service()
-        attr = attributor.get_attributor()
+        cargo_svc = get_cargo_service()
+        package_svc = get_package_service()
+        version_svc = get_version_service()
+        attr = get_attributor()
 
         updater = CargoVersionUpdater(cargo_svc, package_svc, version_svc, attr)
 

src/dagster_app/assets/maven_assets.py

@@ -3,11 +3,11 @@
 
 from dagster import AssetExecutionContext, MetadataValue, Output, asset
 
-from src.dagster_app.resources import (
-    AttributorResource,
-    MavenServiceResource,
-    PackageServiceResource,
-    VersionServiceResource,
+from src.dependencies import (
+    get_attributor,
+    get_maven_service,
+    get_package_service,
+    get_version_service,
 )
 from src.logger import logger
 from src.processes.extractors import MavenPackageExtractor
@@ -22,18 +22,14 @@
 )
 def maven_package_ingestion(
     context: AssetExecutionContext,
-    maven_service: MavenServiceResource,
-    package_service: PackageServiceResource,
-    version_service: VersionServiceResource,
-    attributor: AttributorResource,
 ) -> Output[dict[str, Any]]:
     try:
         logger.info("Starting Maven package ingestion process")
 
-        maven_svc = maven_service.get_service()
-        package_svc = package_service.get_service()
-        version_svc = version_service.get_service()
-        attr = attributor.get_attributor()
+        maven_svc = get_maven_service()
+        package_svc = get_package_service()
+        version_svc = get_version_service()
+        attr = get_attributor()
 
         async def _run():
             new_packages = 0
@@ -131,18 +127,14 @@ async def _run():
 )
 def maven_packages_updates(
     context: AssetExecutionContext,
-    maven_service: MavenServiceResource,
-    package_service: PackageServiceResource,
-    version_service: VersionServiceResource,
-    attributor: AttributorResource,
 ) -> Output[dict[str, Any]]:
     try:
         logger.info("Starting Maven package version update process")
 
-        maven_svc = maven_service.get_service()
-        package_svc = package_service.get_service()
-        version_svc = version_service.get_service()
-        attr = attributor.get_attributor()
+        maven_svc = get_maven_service()
+        package_svc = get_package_service()
+        version_svc = get_version_service()
+        attr = get_attributor()
 
         updater = MavenVersionUpdater(maven_svc, package_svc, version_svc, attr)
 

src/dagster_app/assets/npm_assets.py

@@ -3,11 +3,11 @@
 
 from dagster import AssetExecutionContext, MetadataValue, Output, asset
 
-from src.dagster_app.resources import (
-    AttributorResource,
-    NPMServiceResource,
-    PackageServiceResource,
-    VersionServiceResource,
+from src.dependencies import (
+    get_attributor,
+    get_npm_service,
+    get_package_service,
+    get_version_service,
 )
 from src.logger import logger
 from src.processes.extractors import NPMPackageExtractor
@@ -22,18 +22,14 @@
 )
 def npm_package_ingestion(
     context: AssetExecutionContext,
-    npm_service: NPMServiceResource,
-    package_service: PackageServiceResource,
-    version_service: VersionServiceResource,
-    attributor: AttributorResource,
 ) -> Output[dict[str, Any]]:
     try:
         logger.info("Starting NPM package ingestion process")
 
-        npm_svc = npm_service.get_service()
-        package_svc = package_service.get_service()
-        version_svc = version_service.get_service()
-        attr = attributor.get_attributor()
+        npm_svc = get_npm_service()
+        package_svc = get_package_service()
+        version_svc = get_version_service()
+        attr = get_attributor()
 
         async def _run():
             new_packages = 0
@@ -118,18 +114,14 @@ async def _run():
 )
 def npm_packages_updates(
     context: AssetExecutionContext,
-    npm_service: NPMServiceResource,
-    package_service: PackageServiceResource,
-    version_service: VersionServiceResource,
-    attributor: AttributorResource,
 ) -> Output[dict[str, Any]]:
     try:
         logger.info("Starting NPM package version update process")
 
-        npm_svc = npm_service.get_service()
-        package_svc = package_service.get_service()
-        version_svc = version_service.get_service()
-        attr = attributor.get_attributor()
+        npm_svc = get_npm_service()
+        package_svc = get_package_service()
+        version_svc = get_version_service()
+        attr = get_attributor()
 
         updater = NPMVersionUpdater(npm_svc, package_svc, version_svc, attr)
 

src/dagster_app/assets/nuget_assets.py

@@ -3,11 +3,11 @@
 
 from dagster import AssetExecutionContext, MetadataValue, Output, asset
 
-from src.dagster_app.resources import (
-    AttributorResource,
-    NuGetServiceResource,
-    PackageServiceResource,
-    VersionServiceResource,
+from src.dependencies import (
+    get_attributor,
+    get_nuget_service,
+    get_package_service,
+    get_version_service,
 )
 from src.logger import logger
 from src.processes.extractors import NuGetPackageExtractor
@@ -22,18 +22,14 @@
 )
 def nuget_package_ingestion(
     context: AssetExecutionContext,
-    nuget_service: NuGetServiceResource,
-    package_service: PackageServiceResource,
-    version_service: VersionServiceResource,
-    attributor: AttributorResource,
 ) -> Output[dict[str, Any]]:
     try:
         logger.info("Starting NuGet package ingestion process")
 
-        nuget_svc = nuget_service.get_service()
-        package_svc = package_service.get_service()
-        version_svc = version_service.get_service()
-        attr = attributor.get_attributor()
+        nuget_svc = get_nuget_service()
+        package_svc = get_package_service()
+        version_svc = get_version_service()
+        attr = get_attributor()
 
         async def _run():
             new_packages = 0
@@ -118,18 +114,14 @@ async def _run():
 )
 def nuget_packages_updates(
     context: AssetExecutionContext,
-    nuget_service: NuGetServiceResource,
-    package_service: PackageServiceResource,
-    version_service: VersionServiceResource,
-    attributor: AttributorResource,
 ) -> Output[dict[str, Any]]:
     try:
         logger.info("Starting NuGet package version update process")
 
-        nuget_svc = nuget_service.get_service()
-        package_svc = package_service.get_service()
-        version_svc = version_service.get_service()
-        attr = attributor.get_attributor()
+        nuget_svc = get_nuget_service()
+        package_svc = get_package_service()
+        version_svc = get_version_service()
+        attr = get_attributor()
 
         updater = NuGetVersionUpdater(nuget_svc, package_svc, version_svc, attr)
 

src/dagster_app/assets/pypi_assets.py

@@ -3,11 +3,11 @@
 
 from dagster import AssetExecutionContext, MetadataValue, Output, asset
 
-from src.dagster_app.resources import (
-    AttributorResource,
-    PackageServiceResource,
-    PyPIServiceResource,
-    VersionServiceResource,
+from src.dependencies import (
+    get_attributor,
+    get_package_service,
+    get_pypi_service,
+    get_version_service,
 )
 from src.logger import logger
 from src.processes.extractors import PyPIPackageExtractor
@@ -22,18 +22,14 @@
 )
 def pypi_package_ingestion(
     context: AssetExecutionContext,
-    pypi_service: PyPIServiceResource,
-    package_service: PackageServiceResource,
-    version_service: VersionServiceResource,
-    attributor: AttributorResource,
 ) -> Output[dict[str, Any]]:
     try:
         logger.info("Starting PyPI package ingestion process")
 
-        pypi_svc = pypi_service.get_service()
-        package_svc = package_service.get_service()
-        version_svc = version_service.get_service()
-        attr = attributor.get_attributor()
+        pypi_svc = get_pypi_service()
+        package_svc = get_package_service()
+        version_svc = get_version_service()
+        attr = get_attributor()
 
         async def _run():
             new_packages = 0
@@ -116,18 +112,14 @@ async def _run():
 )
 def pypi_packages_updates(
     context: AssetExecutionContext,
-    pypi_service: PyPIServiceResource,
-    package_service: PackageServiceResource,
-    version_service: VersionServiceResource,
-    attributor: AttributorResource,
 ) -> Output[dict[str, Any]]:
     try:
         logger.info("Starting PyPI package version update process")
 
-        pypi_svc = pypi_service.get_service()
-        package_svc = package_service.get_service()
-        version_svc = version_service.get_service()
-        attr = attributor.get_attributor()
+        pypi_svc = get_pypi_service()
+        package_svc = get_package_service()
+        version_svc = get_version_service()
+        attr = get_attributor()
 
         updater = PyPIVersionUpdater(pypi_svc, package_svc, version_svc, attr)