Merge pull request #5 from Mg30/develop

Develop

Author: Mg30

.github/workflows/contribute.yml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-18.04
     strategy:
       matrix:
-        python-version: [3.8, 3.9, 3.10]
+        python-version: ['3.8', '3.9', '3.10', '3.11']
     steps:
       - uses: actions/checkout@v3
       - name: Set up Python

README.md

@@ -99,7 +99,7 @@ settings.yml
 
 * `project_name/models`: where you will put your tasks
 * `project_name/dags/`: where the corresponding dag PNG file will be
-* `project_name/settings.yml`: a configuration file for your project. This file includes the configuration options for your project, such as the path to your data directory.
+* `settings.yml`: a configuration file for your project. This file includes the configuration options for your project, such as the path to your data directory.
 
 
 
@@ -111,10 +111,20 @@ will export the current state of your dag in the `project_name/dags/` as PNG fil
 
 ## Run your project
 
-`pydwt run`
+`pydwt run <module.function_name>`
 
-will run the current state of your DAG. It will process the tasks in the DAG by level and parrelise
-it with the `ThreadExecutor`
+If no argument provided will run the current state of your DAG. It will process the tasks in the DAG by level and parallelize
+it with the `ThreadExecutor`. It a task failed then its child tasks will not be run.
+
+If argument provided in the form of `module.function_name` for instance `example.task_one` then will run all tasks in the dag leading to this task.  
+If parent tasks succeeded then run the task.
+
+
+## Test your connection setup
+
+`pydwt test-connection`
+
+will test the current setup of your DB connectiona according to your `settings.yml` file.
 
 ## Configuration of your pydwt project
 
@@ -123,12 +133,11 @@ The `settings.yml` file is a configuration file for your pydwt project. It store
 ### connection
 The connection section contains the configuration details for connecting to the database. The available options are:
 
-* `db`: the name of the database
-* `host`: the hostname or IP address of the database server
-* `user`: the username for the database connection
-* `password`: the password for the database connection
-* `port`: the port number to use for the database connection
-* `sql_alchemy_driver`: the SQLAlchemy driver to use for the database connection
+* `url`: the connection string to your db
+
+You can add others keys that will be forwarded to the underlying `create_engine` function
+for instance you can add a `echo : true` and it will call `create_engine(url=url, echo=echo)`
+see [here](https://docs.sqlalchemy.org/en/20/core/engines.html#sqlalchemy.create_engine) supported args.
 
 ### project
 The project section contains the project-related settings. The available options are:

poetry.lock

@@ -1,12 +1,13 @@
+import traceback
 from typing import Dict
 import os
 import sys
-
+from typing import Optional
 import typer
 import yaml
 from dependency_injector.wiring import register_loader_containers
-
 from pydwt.core.containers import Container
+import logging
 
 
 sys.path.append(os.getcwd())
@@ -21,6 +22,7 @@ def load_config(path: str) -> Dict:
         config = yaml.safe_load(f)
     return config
 
+
 # Define command-line interface using Typer
 @app.command()
 def new(project_name: str):
@@ -30,12 +32,12 @@ def new(project_name: str):
 
 
 @app.command()
-def run():
+def run(name: Optional[str] = typer.Argument(None)):
     """Run the workflow DAG for the current project."""
     config = load_config(path="settings.yml")
     container.config.from_dict(config)
     project_handler = container.project_factory()
-    project_handler.run()
+    project_handler.run(name)
 
 
 @app.command()
@@ -47,6 +49,21 @@ def export_dag():
     project_handler.export_dag()
 
 
+@app.command()
+def test_connection():
+    """Export the workflow DAG for the current project."""
+    config = load_config(path="settings.yml")
+    container.config.from_dict(config)
+    try:
+        conn = container.database_client()
+        engine = conn.get_engine()
+        dbapi = engine.connect()
+        dbapi.close()
+        logging.info("successfully connected to db")
+    except Exception:
+        logging.error(f"connection failed {traceback.print_exc()}")
+
+
 if __name__ == "__main__":
     # Run the command-line interface
     try:

pydwt/app.py

@@ -1,38 +1,19 @@
 from dataclasses import dataclass, field
+from typing import Dict
 from sqlalchemy import create_engine, Engine
-from urllib.parse import quote_plus
 
 
 @dataclass
 class Connection(object):
     """Class representing a connection to a database.
 
     Attributes:
-        db (str): Name of the database.
-        host (int): Hostname or IP address of the server.
-        name (str): Name of the connection.
-        password (str): Password for the connection.
-        user (str): Username for the connection.
-        sql_alchemy_driver (str): SQL Alchemy driver.
+        params (Dict): containning SQL alchemy DB url and Kwargs forwarded to create_engine
         engine (sqlalchemy.engine.Engine): Database engine.
     """
 
-    db: str
-    host: int
-    name: str
-    password: str
-    user: str
-    port: int
-    sql_alchemy_driver: str
+    params: Dict
     engine: Engine = field(init=False, default=None)
 
-    def __post_init__(self):
-        """Initialize the connection object and create the database engine."""
-        # Encode the password to be used in the connection URL
-        self.password = quote_plus(self.password)
-        # Create the database engine using SQLAlchemy
-
     def get_engine(self):
-        return create_engine(
-            f"{self.sql_alchemy_driver}://{self.user}:{self.password}@{self.host}:{self.port}/{self.db}"
-        )
+        return create_engine(**self.params)

pydwt/context/connection.py

@@ -24,13 +24,7 @@ class Container(containers.DeclarativeContainer):
     # Singleton provider that provides the database connection instance
     database_client = providers.ThreadSafeSingleton(
         Connection,
-        config.connection.db,
-        config.connection.host,
-        config.connection.name,
-        config.connection.password,
-        config.connection.user,
-        config.connection.port,
-        config.connection.sql_alchemy_driver,
+        config.connection,
     )
 
     # Singleton provider that provides the datasources instance

pydwt/core/containers.py

@@ -16,20 +16,9 @@ def __init__(self, tasks: List) -> None:
         self.tasks = tasks
         self.graph = nx.DiGraph()
         self.source = "s"
-        self.nodes_by_level = {}
         self.node_index = {}
         self.node_names = {}
 
-    @property
-    def levels(self) -> Dict[int, List[int]]:
-        """Get the level of each task in the dag.
-
-        Returns:
-            Dict[int, List[int]]: Dictionary of levels and their corresponding task indexes.
-        """
-        self.build_level()
-        return self.nodes_by_level
-
     def build_dag(self) -> None:
         """Build the directed acyclic graph from the tasks and their dependencies."""
         edges = []
@@ -59,17 +48,40 @@ def build_dag(self) -> None:
 
         self.graph.add_edges_from(edges)
 
-    def build_level(self) -> None:
-        """Assign levels to nodes based on the breadth-first search."""
+    def build_level(self, target: str = None) -> Dict:
+        """Assign levels to nodes in the dag using the breadth-first search.
+
+        Args:
+            target (str): Optional target node. If provided, the search will be performed up to this node.
+
+        Returns:
+            Dict: Dictionary of levels and their corresponding node indexes.
+
+        Raises:
+            KeyError: If target node is not found in the graph.
+        """
+        nodes_by_level = {}
+        # If a target node is provided, get its index in the node_index dictionary
+        node_index = self.node_index.get(target, None)
         # Perform breadth-first search on the graph
-        self.nodes_by_level = {}
         bfs_tree = nx.bfs_tree(self.graph, self.source)
+
+        level = None
+
+        if node_index is not None:
+            level = {}
+            path = nx.shortest_path(self.graph, self.source, node_index)
+            for i,node in enumerate(path):
+                level[node] = i
+
+        else:
         # Assign levels to nodes based on their distance from the root node
-        level = nx.shortest_path_length(bfs_tree, self.source)
+            level = nx.shortest_path_length(bfs_tree, self.source)
         for node, node_level in level.items():
-            if node_level not in self.nodes_by_level:
-                self.nodes_by_level[node_level] = []
-            self.nodes_by_level[node_level].append(node)
+            if node_level not in nodes_by_level:
+                nodes_by_level[node_level] = []
+            nodes_by_level[node_level].append(node)
+        return nodes_by_level
 
     def check_parents_status(self, task):
         """Check if all parent tasks have the attribute status set to success.
@@ -82,7 +94,9 @@ def check_parents_status(self, task):
         """
         node_index = self.node_index[task.name]
         for parent_index in self.graph.predecessors(node_index):
+            if parent_index == "s":
+                continue
             parent = self.tasks[parent_index]
-            if parent.status == Status.ERROR:
+            if parent.status == Status.ERROR or parent.status == Status.PENDING:
                 return False
         return True

pydwt/core/dag.py

@@ -62,11 +62,12 @@ def import_all_models(self) -> None:
         for model in models:
             importlib.import_module(f"{self.name}.models.{model}")
 
-    def run(self) -> None:
+    def run(self, task_name: str = None) -> None:
         """Run the DAG-based workflow."""
+        task_full_name = f"{self.name}.models.{task_name}" if task_name else None
         self.import_all_models()
         self.workflow.dag.build_dag()
-        self.workflow.run()
+        self.workflow.run(task_full_name)
 
     def export_dag(self) -> None:
         """Export the DAG to a PNG image file."""
@@ -115,12 +116,8 @@ def _create_settings(self, project_name: str) -> None:
             "tasks": {"task_one": {"materialize": "view"}},
             "sources": {"one": {"table": "table_name", "schema": "some_schema"}},
             "connection": {
-                "db": "",
-                "host": "",
-                "port": 0,
-                "password": "",
-                "user": "",
-                "sql_alchemy_driver": "",
+            "url": '<connection-string>',
+            "echo": True
             },
         }
         if not os.path.exists(settings_projects):

pydwt/core/project.py

@@ -29,10 +29,10 @@ def __post_init__(self) -> None:
         # check if cache strategy is provided and use_cache flag is enabled
         self.dag = Dag(self.tasks)
 
-    def run(self) -> None:
+    def run(self, task_name: str = None) -> None:
         """Run the tasks in the DAG."""
         # Get a dictionary of all the task levels in the DAG
-        levels = self.dag.levels
+        levels = self.dag.build_level(task_name)
         start_time_workflow = time.time()
         # Iterate through each level in the DAG
         for level, task_indexes in levels.items():

pydwt/core/workflow.py

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pydwt"
-version = "0.1.1a1"
+version = "0.1.1"
 description = ""
 authors = ["gonza <[email protected]>"]
 readme = "README.md"
@@ -10,7 +10,6 @@ python = "^3.8"
 networkx = "^3.0"
 matplotlib = "^3.6.3"
 sqlalchemy = "^2.0.1"
-dill = "^0.3.6"
 typer = "^0.7.0"
 pyyaml = "^6.0"
 dependency-injector = "^4.41.0"