Can now export a NetworkX graph with attributes and version bump

Author: JackAtOmenApps

django_postgresql_dag/exceptions.py

@@ -8,3 +8,19 @@ class NodeNotReachableException(Exception):
 
     pass
 
+
+class GraphModelsCannotBeParsedException(Exception):
+    """
+    Exception for node distance and path
+    """
+
+    pass
+
+
+class IncorrectUsageException(Exception):
+    """
+    Exception for node distance and path
+    """
+
+    pass
+

django_postgresql_dag/transformations.py

@@ -3,10 +3,16 @@
 django-postgresql-dag to alternate formats.
 """
 
+from django.core.exceptions import FieldDoesNotExist
 from django.db.models import Case, When
+from django.db.models.fields import DateTimeField, UUIDField
+from django.db.models.fields.files import ImageField, FileField
+from django.db.models.fields.related import ManyToManyField
 
+from .exceptions import GraphModelsCannotBeParsedException, IncorrectUsageException
+
+from itertools import chain
 import networkx as nx
-import pandas as pd
 
 
 def _filter_order(queryset, field_names, values):
@@ -30,54 +36,184 @@ def _filter_order(queryset, field_names, values):
     return queryset.filter(**filter_condition).order_by(order_by)
 
 
-def rawqueryset_to_values_list(rawqueryset):
-    """Returns a list of lists of each instance"""
-    columns = rawqueryset.columns
-    for row in rawqueryset:
-        yield tuple(getattr(row, column) for column in columns)
+def get_queryset_characteristics(queryset):
+    """
+    Returns a tuple of the node & edge model classes and the queryset type
+    for the provided queryset
+    """
+    try:
+        # Assume a queryset of nodes was provided
+        _NodeModel = queryset.model
+        _EdgeModel = queryset.model._meta.get_field("parents").through
+        queryset_type = "nodes_queryset"
+    except FieldDoesNotExist:
+        try:
+            # Assume a queryset of edges was provided
+            _EdgeModel = queryset.model
+            _NodeModel = queryset.model._meta.get_field("parent").related_model
+            queryset_type = "edges_queryset"
+        except FieldDoesNotExist:
+            raise GraphModelsCannotBeParsedException
+    return (_NodeModel, _EdgeModel, queryset_type)
+
+
+def model_to_dict(instance, fields=None, date_strf=None):
+    """
+    Returns a dictionary of {field_name: field_value} for a given model instance
+    e.g.: model_to_dict(myqueryset.first(), fields=["id",])
+
+    For DateTimeFields, a formatting string can be provided
 
+    Adapted from: https://ziwon.github.io/post/using_custom_model_to_dict_in_django/
+    """
 
-def rawqueryset_to_dataframe(rawqueryset):
-    """Returns a pandas dataframe"""
-    return pd.DataFrame(
-        rawqueryset_to_values_list(rawqueryset), columns=list(rawqueryset.columns)
-    )
+    if not fields:
+        raise IncorrectUsageException("fields list must be provided")
+
+    opts = instance._meta
+    data = {}
+    __fields = list(map(lambda a: a.split("__")[0], fields or []))
+
+    for f in chain(opts.concrete_fields, opts.private_fields, opts.many_to_many):
+        is_editable = getattr(f, "editable", False)
+
+        if fields and f.name not in __fields:
+            continue
+
+        if isinstance(f, DateTimeField):
+            dt = f.value_from_object(instance)
+            # Format based on format string provided, otherwise return a timestamp
+            data[f.name] = dt.strftime(date_strf) if date_strf else dt.timestamp()
+
+        elif isinstance(f, ImageField):
+            image = f.value_from_object(instance)
+            data[f.name] = image.url if image else None
+
+        elif isinstance(f, FileField):
+            file = f.value_from_object(instance)
+            data[f.name] = file.url if file else None
+
+        elif isinstance(f, ManyToManyField):
+            if instance.pk is None:
+                data[f.name] = []
+            else:
+                qs = f.value_from_object(instance)
+                if qs._result_cache is not None:
+                    data[f.name] = [item.pk for item in qs]
+                else:
+                    try:
+                        m2m_field = list(
+                            filter(lambda a: f.name in a and a.find("__") != -1, fields)
+                        )[0]
+                        key = m2m_field[len(f.name) + 2 :]
+                        data[f.name] = list(qs.values_list(key, flat=True))
+                    except IndexError:
+                        data[f.name] = list(qs.values_list("pk", flat=True))
+
+        if isinstance(f, UUIDField):
+            uuid = f.value_from_object(instance)
+            data[f.name] = str(uuid) if uuid else None
+
+        # ToDo: Process other model fields
+
+        elif is_editable:
+            data[f.name] = f.value_from_object(instance)
+
+    funcs = set(__fields) - set(list(data.keys()))
+    for func in funcs:
+        obj = getattr(instance, func)
+        if inspect.ismethod(obj):
+            data[func] = obj()
+        else:
+            data[func] = obj
+    return data
+
+
+def edges_from_nodes_queryset(nodes_queryset):
+    """Given an Edge Model and a QuerySet or RawQuerySet of nodes,
+    returns a queryset of the associated edges"""
+    _NodeModel, _EdgeModel, queryset_type = get_queryset_characteristics(nodes_queryset)
+
+    if queryset_type == "nodes_queryset":
+        return _filter_order(_EdgeModel.objects, ["parent", "child"], nodes_queryset)
+    raise IncorrectQuerysetTypeException
+
+
+def nodes_from_edges_queryset(edges_queryset):
+    """Given a Node Model and a QuerySet or RawQuerySet of edges,
+    returns a queryset of the associated nodes"""
+    _NodeModel, _EdgeModel, queryset_type = get_queryset_characteristics(edges_queryset)
+
+    if queryset_type == "edges_queryset":
+
+        nodes_list = (
+            _filter_order(
+                _NodeModel.objects,
+                [
+                    f"{_NodeModel.__name__}_child",
+                ],
+                edges_queryset,
+            )
+            | _filter_order(
+                _NodeModel.objects,
+                [
+                    f"{_NodeModel.__name__}_parent",
+                ],
+                edges_queryset,
+            )
+        ).values_list("pk")
+
+        return _NodeModel.objects.filter(pk__in=nodes_list)
+    raise IncorrectQuerysetTypeException
+
+
+def nx_from_queryset(
+    queryset,
+    graph_attributes_dict=None,
+    node_attribute_fields_list=None,
+    edge_attribute_fields_list=None,
+    date_strf=None,
+):
+    """
+    Provided a queryset of nodes or edges, returns a NetworkX graph
 
+    Optionally, the following can be supplied to add attributes to components of the generated graph:
+    graph_attributes_dict: A dictionary of attributes to add to the graph itself
+    node_attribute_fields_list: a list of strings of field names to be added to nodes
+    edge_attribute_fields_list: a list of strings of field names to be added to edges
+    """
+    _NodeModel, _EdgeModel, queryset_type = get_queryset_characteristics(queryset)
 
-def edges_from_nodes_queryset(edge_model, nodes_queryset):
-    """Given an Edge Model and a QuerySet or RawQuerySet of nodes, returns a queryset of the associated edges"""
-    return _filter_order(edge_model.objects, ["parent", "child"], nodes_queryset)
+    if graph_attributes_dict is None:
+        graph_attributes_dict = {}
 
+    graph = nx.Graph(**graph_attributes_dict)
 
-def nodes_from_edges_queryset(node_model, edges_queryset):
-    """Given a Node Model and a QuerySet or RawQuerySet of edges, returns a queryset of the associated nodes"""
-    nodes_list = (
-        _filter_order(
-            node_model.objects,
-            [
-                f"{node_model.__name__}_child",
-            ],
-            edges_queryset,
-        )
-        | _filter_order(
-            node_model.objects,
-            [
-                f"{node_model.__name__}_parent",
-            ],
-            edges_queryset,
-        )
-    ).values_list("pk")
+    if queryset_type == "nodes_queryset":
+        nodes_queryset = queryset
+        edges_queryset = edges_from_nodes_queryset(nodes_queryset)
+    else:
+        edges_queryset = queryset
+        nodes_queryset = nodes_from_edges_queryset(edges_queryset)
 
-    return node_model.objects.filter(pk__in=nodes_list)
+    for node in nodes_queryset:
+        if node_attribute_fields_list is not None:
+            node_attribute_fields_dict = model_to_dict(
+                node, fields=node_attribute_fields_list, date_strf=date_strf
+            )
+        else:
+            node_attribute_fields_dict = {}
 
+        graph.add_node(node.pk, **node_attribute_fields_dict)
 
-def nx_from_nodes_queryset(nodes_queryset):
-    """Provided a queryset of nodes, returns a NetworkX graph"""
-    # ToDo: Implement
-    pass
+    for edge in edges_queryset:
+        if edge_attribute_fields_list is not None:
+            edge_attribute_fields_dict = model_to_dict(
+                edge, fields=edge_attribute_fields_list, date_strf=date_strf
+            )
+        else:
+            edge_attribute_fields_dict = {}
 
+        graph.add_edge(edge.parent.pk, edge.child.pk, **edge_attribute_fields_dict)
 
-def nx_from_edges_queryset(edges_queryset, fields_array=None):
-    """Provided a queryset of edges, returns a NetworkX graph"""
-    # ToDo: Implement
-    graph = nx.Graph()
+    return graph

setup.py

@@ -3,7 +3,7 @@
 import os
 from setuptools import setup
 
-version = '0.0.18'
+version = '0.0.19'
 
 classifiers = [
     "Development Status :: 3 - Alpha",