infeeeee/dyn2py
1
0
Fork 0
mirror of https://github.com/infeeeee/dyn2py synced 2025-12-16 22:16:18 +01:00
Code Issues Packages Projects Releases Activity

Rework exceptions, class methods for open files

Browse Source
This commit is contained in:
Péter Gyetvai
2023-04-12 07:15:20 +02:00
parent 5efae02594
commit 72cb52e0bf
8 changed files with 124 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
README.md
View File

@@ -109,8 +109,8 @@ Most basic example to extract all nodes next to a Dynamo file:
import dyn2py
dynamo_file = dyn2py.DynamoFile("path/to/dynamofile.dyn")
python_files = dynamo_file.extract_python()
[python_file.write() for python_file in python_files]
dynamo_file.extract_python()
dyn2py.PythonFile.write_open_files()
```
Change options like with the command line switches with the `Options` class:

27
dyn2py/__init__.py
View File

@@ -22,11 +22,7 @@ __all__ = [
"File",
"DynamoFile",
"PythonFile",
"PythonNode",
"DynamoFileException",
"PythonNodeNotFoundException",
"PythonNodeException",
"PythonFileException"
"PythonNode"
]
@@ -155,13 +151,13 @@ def run(options: Options) -> None:
for f in source_files:
try:
files.append(File(f))
except DynamoFileException as e:
except DynamoFile.Error as e:
# It's a dynamo1 file
logging.warning(e)
logging.warning(f"This is a Dynamo 1 file! {e.file.filepath}")
continue
except PythonNodeNotFoundException as e:
# No python node in this file
logging.warning(e)
except DynamoFile.PythonNodeNotFound as e:
# No python nodes in this file
logging.warning(f"This file has no Python nodes! {e.file.filepath} ")
continue
# Dynamo files come first, sort sources:
@@ -202,11 +198,10 @@ def run(options: Options) -> None:
try:
f.update_dynamo(options)
except FileNotFoundError:
logging.error(f"Source Dynamo file not found! {f.filepath}")
logging.error(f"{f.filepath} Source Dynamo file not found! ")
# Write files at the end:
for f in DynamoFile.open_files | PythonFile.open_files:
try:
f.write(options)
except FileNotFoundError:
logging.error(f"Cannot save file! {f.filepath}")
try:
File.write_open_files()
except File.Error as e:
logging.error(f"Cannot save file! {e.file.filepath}")

16
dyn2py/exceptions.py
View File

@@ -1,16 +0,0 @@
class DynamoFileException(Exception):
"""Something wrong in this DynamoFile"""
pass
class PythonNodeNotFoundException(Exception):
"""PythonNode not found"""
pass
class PythonNodeException(Exception):
pass
class PythonFileException(Exception):
pass

116
dyn2py/files.py
View File

@@ -9,7 +9,6 @@ from decimal import Decimal
from pathvalidate import sanitize_filename
from importlib_metadata import metadata
from dyn2py.exceptions import *
from dyn2py.options import Options
@@ -20,6 +19,9 @@ HEADER_SEPARATOR = "*" * 60
class File():
"""Base class for managing files"""
open_files: set[File] = set()
"""A set of open files."""
def __init__(self, filepath: pathlib.Path | str, read_from_disk: bool = True) -> None:
"""Generate a file object. If the path is correct it will become a DynamoFile or PythonFile object.
Calls DynamoFile.read_file() and PythonFile.read_file()
@@ -120,7 +122,7 @@ class File():
Raises:
TypeError: If called on a File object
FileNotFoundError: Target folder does not exist
File.Error: Target folder does not exist
"""
if not options:
@@ -150,7 +152,7 @@ class File():
f"Should write file, but it's a dry-run: {self.filepath}")
else:
if not self.dirpath.exists():
raise FileNotFoundError("File dir does not exist!")
raise File.Error("File dir does not exist!", self)
logging.info(f"Writing file: {self.filepath}")
self._write_file()
if options.loglevel == "HEADLESS":
@@ -165,6 +167,45 @@ class File():
raise NotImplementedError(
"Should be called only on DynamoFile and PythonFile objects!")
@classmethod
def get_open_files(cls) -> set:
"""Get open files of this class and subclasses
Returns:
set: A set of open files
"""
return {f for f in File.open_files if
isinstance(f, cls)}
@classmethod
def write_open_files(cls, options: Options | None = None) -> None:
"""Write open files of this class and subclasses
Args:
options (Options | None, optional): Run options. Defaults to None.
"""
if not options:
options = Options()
for f in cls.get_open_files():
f.write(options)
@classmethod
def close_open_files(cls) -> None:
"""Close open files of this class and subclasses"""
File.open_files = File.open_files - cls.get_open_files()
class Error(Exception):
def __init__(self, message: str, file: File) -> None:
"""There is some problem with this file
Args:
message (str): The message to display
file (File): The problem File
"""
super().__init__(message)
self.file = file
class DynamoFile(File):
"""A Dynamo file, subclass of File()"""
@@ -178,9 +219,6 @@ class DynamoFile(File):
python_nodes: set[PythonNode]
"""Python node objects, read from this file."""
open_files: set[DynamoFile] = set()
"""A set of open Dynamo files, before saving. Self added by read()"""
def extract_python(self, options: Options | None = None) -> list[PythonFile]:
"""Extract python files from Dynamo graphs, add them to open_files
@@ -229,9 +267,9 @@ class DynamoFile(File):
Raises:
FileNotFoundError: The file does not exist
DynamoFileException: If the file is a Dynamo 1 file
DynamoFile.Error: If the file is a Dynamo 1 file
json.JSONDecodeError: If there are any other problem with the file
PythonNodeNotFoundException: No python nodes in the file
DynamoFile.PythonNodeNotFound: No python nodes in the file
"""
if not self.exists:
@@ -249,7 +287,7 @@ class DynamoFile(File):
except json.JSONDecodeError as e:
with open(self.filepath, "r", encoding="utf-8") as input_json:
if input_json.readline().startswith("<Workspace Version="):
raise DynamoFileException("This is a Dynamo 1 file!")
raise self.Error("This is a Dynamo 1 file!", self)
else:
raise e
@@ -264,8 +302,8 @@ class DynamoFile(File):
node_views = self.full_dict["View"]["NodeViews"]
if not full_python_nodes:
raise PythonNodeNotFoundException(
"No python nodes in this file!")
raise self.PythonNodeNotFound(
"No python nodes in this file!", self, "")
self.python_nodes = set()
@@ -286,7 +324,7 @@ class DynamoFile(File):
PythonNode: The PythonNode with the given id
Raises:
PythonNodeNotFoundException: No python node with this id
DynamoFile.PythonNodeNotFound: No python node with this id
"""
python_node = next((
@@ -294,8 +332,8 @@ class DynamoFile(File):
), None)
if not python_node:
raise PythonNodeNotFoundException(
f"Node not found with id {node_id}")
raise self.PythonNodeNotFound(
"Node not found", self, node_id)
return python_node
@@ -306,7 +344,7 @@ class DynamoFile(File):
python_node (PythonNode): The new node
Raises:
PythonNodeNotFoundException: Existing node not found
DynamoFile.PythonNodeNotFound: Existing node not found
"""
# Find the old node:
@@ -316,8 +354,9 @@ class DynamoFile(File):
n for n in self.full_dict["Nodes"] if n["Id"] == python_node.id
), {})
if not node_dict or not python_node_in_file:
raise PythonNodeNotFoundException()
if not node_dict:
raise self.PythonNodeNotFound(
"Existing node not found in file", self, python_node.id)
# Remove the old and add the new:
self.python_nodes.remove(python_node_in_file)
@@ -368,11 +407,24 @@ class DynamoFile(File):
Returns:
DynamoFile: The file. None if not found
"""
f = next((d for d in DynamoFile.open_files if d.uuid == uuid), None)
f = next((d for d in DynamoFile.get_open_files() if d.uuid == uuid), None)
if f:
logging.debug(f"Found open file {f.uuid}")
return f
class PythonNodeNotFound(Exception):
def __init__(self, message: str, file: DynamoFile, node_id: str) -> None:
"""Python node not found with this id
Args:
message (str): The message to display
file (DynamoFile): The problem DynamoFile
node_id (str): The missing id
"""
super().__init__(message)
self.file = file
self.node_id = node_id
class PythonFile(File):
"""A Python file, subclass of File()"""
@@ -384,9 +436,6 @@ class PythonFile(File):
text: str
"""Full contents of the file."""
open_files: set[PythonFile] = set()
"""A set of open Python files."""
def __init__(self,
filepath: pathlib.Path | str,
dynamo_file: DynamoFile | None = None,
@@ -411,7 +460,8 @@ class PythonFile(File):
])
# Calculate relative path, change to forward slash
dyn_path_string = os.path.relpath(dynamo_file.filepath, self.dirpath)
dyn_path_string = os.path.relpath(
dynamo_file.filepath, self.dirpath)
if "\\" in dyn_path_string:
dyn_path_string = dyn_path_string.replace("\\", "/")
@@ -458,7 +508,7 @@ class PythonFile(File):
Raises:
FileNotFoundError: The file does not exist
PythonFileException: Some error reading the file
PythonFile.Error: Some error reading the file
"""
if not self.exists:
raise FileNotFoundError
@@ -493,7 +543,7 @@ class PythonFile(File):
# Find the location of the separator
sl = line.find(":")
if sl == -1:
raise PythonFileException("Error reading header!")
raise self.Error("Error reading header!", self)
self.header_data[line[0:sl]] = line[sl+1:]
self.code = python_lines[code_start_line:]
@@ -535,7 +585,7 @@ class PythonFile(File):
"""Get the source Dynamo file of this PythonFile
Raises:
DynamoFileException: The uuid of the dynamo file changed
DynamoFile.Error: The uuid of the dynamo file changed
Returns:
DynamoFile: The DynamoFile
@@ -561,7 +611,8 @@ class PythonFile(File):
# Check if uuid is ok:
if not dynamo_file.uuid == self.header_data["dyn_uuid"]:
raise DynamoFileException(f"Dynamo graph uuid changed!")
raise DynamoFile.Error(
"Dynamo graph uuid changed!", dynamo_file)
return dynamo_file
@@ -602,7 +653,7 @@ class PythonNode():
python_file (PythonFile, optional): The python file to be converted to node. Defaults to None.
Raises:
PythonNodeException: Wrong arguments were given
Error: Wrong arguments were given
"""
# Initialize from dynamo file:
if node_dict_from_dyn and dynamo_file and not python_file:
@@ -642,9 +693,18 @@ class PythonNode():
self.filename = python_file.basename + ".py"
self.filepath = python_file.filepath
elif python_file and node_dict_from_dyn and dynamo_file:
raise self.Error("Too much arguments given!")
elif not python_file and not node_dict_from_dyn and not dynamo_file:
raise self.Error("No arguments given!")
else:
raise PythonNodeException
raise self.Error("Something wrong!")
# Calculate checksum:
checksums = [hashlib.md5(l.encode()).hexdigest() for l in self.code]
self.checksum = hashlib.md5("".join(checksums).encode()).hexdigest()
class Error(Exception):
pass

3
tests/support.py
View File

@@ -31,8 +31,7 @@ def extract_single_node_dyn(modify_py: bool = False):
pythonfiles = dyn.extract_python(options)
pythonfiles[0].write()
dyn2py.PythonFile.open_files.clear()
dyn2py.DynamoFile.open_files.clear()
dyn2py.File.open_files.clear()
if modify_py:
# Open the extracted file and replace a string:

13
tests/test_DynamoFile.py
View File

@@ -19,7 +19,7 @@ class TestDynamoFile(unittest.TestCase):
self.assertTrue(dyn in dyn2py.DynamoFile.open_files)
# Dynamo 1 file:
with self.assertRaises(dyn2py.DynamoFileException):
with self.assertRaises(dyn2py.DynamoFile.Error):
dyn1 = dyn2py.DynamoFile(f"{INPUT_DIR}/dynamo1file.dyn")
# Not existing file:
@@ -27,7 +27,7 @@ class TestDynamoFile(unittest.TestCase):
dyn2 = dyn2py.DynamoFile(f"{INPUT_DIR}/not_existing.dyn")
# No python nodes:
with self.assertRaises(dyn2py.PythonNodeNotFoundException):
with self.assertRaises(dyn2py.DynamoFile.PythonNodeNotFound):
dyn2 = dyn2py.DynamoFile(f"{INPUT_DIR}/no_python.dyn")
def test_get_python_nodes(self):
@@ -39,7 +39,7 @@ class TestDynamoFile(unittest.TestCase):
self.assertIn(py_node, dyn.python_nodes)
self.assertEqual(py_node.checksum, "e830a6ae6b395bcfd4e5a40da48f3bfc")
with self.assertRaises(dyn2py.PythonNodeNotFoundException):
with self.assertRaises(dyn2py.DynamoFile.PythonNodeNotFound):
dyn.get_python_node_by_id("wrongid")
def test_extract_python(self):
@@ -50,10 +50,9 @@ class TestDynamoFile(unittest.TestCase):
dyn = dyn2py.DynamoFile(f"{INPUT_DIR}/python_nodes.dyn")
dyn.extract_python(options=opt)
self.assertEqual(len(dyn2py.PythonFile.open_files), 6)
self.assertEqual(len(dyn2py.PythonFile.get_open_files()), 6)
for f in dyn2py.PythonFile.open_files:
f.write()
dyn2py.PythonFile.write_open_files()
output_dir = pathlib.Path(OUTPUT_DIR)
self.assertEqual(len(list(output_dir.iterdir())), 6)
@@ -139,6 +138,6 @@ class TestDynamoFile(unittest.TestCase):
self.assertTrue(node2)
self.assertEqual(node1.checksum, node2.checksum)
with self.assertRaises(dyn2py.PythonNodeNotFoundException):
with self.assertRaises(dyn2py.DynamoFile.PythonNodeNotFound):
node2.id = "wrong_id"
dyn2.update_python_node(node2)

22
tests/test_PythonFile.py
View File

@@ -12,9 +12,10 @@ class TestPythonFile(unittest.TestCase):
def test_init(self):
extract_single_node_dyn()
py1 = dyn2py.PythonFile(f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py")
dyn2py.File.open_files.clear()
dyn2py.DynamoFile.open_files.clear()
py1 = dyn2py.PythonFile(
f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py")
dyn = dyn2py.DynamoFile(f"{INPUT_DIR}/single_node.dyn")
node = list(dyn.python_nodes)[0]
py2 = dyn2py.PythonFile(filepath=node.filepath,
@@ -23,9 +24,10 @@ class TestPythonFile(unittest.TestCase):
for py in [py1, py2]:
self.assertEqual(len(py.code), 17)
self.assertEqual(len(py.text.split(os.linesep)), 32, msg=py.filepath)
self.assertEqual(len(py.text.split(os.linesep)),
32, msg=py.filepath)
self.assertIs(type(py.header_data), dict)
self.assertTrue(py in dyn2py.PythonFile.open_files)
self.assertTrue(py in dyn2py.PythonFile.get_open_files())
def test_update_dynamo(self):
extract_single_node_dyn(modify_py=True)
@@ -68,14 +70,13 @@ class TestPythonFile(unittest.TestCase):
def test_get_source_dynamo_file(self):
extract_single_node_dyn()
dyn2py.DynamoFile.open_files.clear()
dyn2py.PythonFile.open_files.clear()
dyn2py.File.open_files.clear()
py1 = dyn2py.PythonFile(
f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py")
dyn1 = py1.get_source_dynamo_file()
self.assertEqual(len(dyn2py.DynamoFile.open_files), 1)
self.assertEqual(len(dyn2py.DynamoFile.get_open_files()), 1)
self.assertIn(dyn1, dyn2py.DynamoFile.open_files)
dyn2 = py1.get_source_dynamo_file()
@@ -83,14 +84,13 @@ class TestPythonFile(unittest.TestCase):
dyn2py.DynamoFile.open_files.clear()
with self.assertRaises(dyn2py.DynamoFileException):
with self.assertRaises(dyn2py.DynamoFile.Error):
py1.header_data["dyn_uuid"] = "wrong-uuid"
py1.get_source_dynamo_file()
def test_write(self):
extract_single_node_dyn()
dyn2py.DynamoFile.open_files.clear()
dyn2py.PythonFile.open_files.clear()
dyn2py.File.open_files.clear()
py1 = dyn2py.PythonFile(
f"{OUTPUT_DIR}/single_node_1c5d99792882409e97e132b3e9f814b0.py")
@@ -98,7 +98,7 @@ class TestPythonFile(unittest.TestCase):
dyn1 = py1.get_source_dynamo_file()
node = list(dyn1.python_nodes)[0]
py2 = dyn2py.PythonFile(
node.filepath, dynamo_file=dyn1, python_node=node)
f"{OUTPUT_DIR}/{node.filename}", dynamo_file=dyn1, python_node=node)
self.assertIsNot(py1, py2)
self.assertEqual(py1.code, py2.code)
for d in py1.header_data:

3
tests/test_PythonNode.py
View File

@@ -48,11 +48,12 @@ class TestPythonNode(unittest.TestCase):
py = dyn2py.PythonFile(f"{OUTPUT_DIR}/single_node_mod.py")
with self.assertRaises(dyn2py.PythonNodeException):
with self.assertRaises(dyn2py.PythonNode.Error):
node1 = dyn2py.PythonNode(
node_dict_from_dyn=node_dict,
dynamo_file=dyn,
python_file=py
)
with self.assertRaises(dyn2py.PythonNode.Error):
node2 = dyn2py.PythonNode()