Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add poster method that raises an exception #1038

Merged
merged 5 commits into from
Jan 13, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions doc/changelog.d/1038.added.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
add poster method that raises an exception
34 changes: 33 additions & 1 deletion src/ansys/mechanical/core/embedding/poster.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,19 @@
import typing


class PosterError(Exception):
"""Class which holds errors from the background thread posting system."""

def __init__(self, error: Exception):
"""Create an instance to hold the given error."""
self._error = error

Check warning on line 33 in src/ansys/mechanical/core/embedding/poster.py

View check run for this annotation

Codecov / codecov/patch

src/ansys/mechanical/core/embedding/poster.py#L33

Added line #L33 was not covered by tests

@property
def error(self) -> Exception:
"""Get the underlying exception."""
return self._error

Check warning on line 38 in src/ansys/mechanical/core/embedding/poster.py

View check run for this annotation

Codecov / codecov/patch

src/ansys/mechanical/core/embedding/poster.py#L38

Added line #L38 was not covered by tests


class Poster:
"""Class which can post a python callable function to Mechanical's main thread."""

Expand All @@ -37,7 +50,26 @@

self._poster = Ans.Common.WB1ManagedUtils.TaskPoster

def post(self, callable: typing.Callable):
def try_post(self, callable: typing.Callable) -> typing.Any:
"""Post the callable to Mechanical's main thread.

This does the same thing as `post` but if `callable`
raises an exception, try_post will raise the same
exception to the caller of `try_post`.
"""

def wrapped():
try:
return callable()
except Exception as e:
return PosterError(e)

Check warning on line 65 in src/ansys/mechanical/core/embedding/poster.py

View check run for this annotation

Codecov / codecov/patch

src/ansys/mechanical/core/embedding/poster.py#L64-L65

Added lines #L64 - L65 were not covered by tests

result = self.post(wrapped)
if isinstance(result, PosterError):
raise result.error

Check warning on line 69 in src/ansys/mechanical/core/embedding/poster.py

View check run for this annotation

Codecov / codecov/patch

src/ansys/mechanical/core/embedding/poster.py#L69

Added line #L69 was not covered by tests
return result

def post(self, callable: typing.Callable) -> typing.Any:
"""Post the callable to Mechanical's main thread.

The main thread needs to be receiving posted messages
Expand Down
15 changes: 14 additions & 1 deletion tests/embedding/test_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,7 @@ def test_app_poster(embedded_app):
poster = embedded_app.poster

name = []
error = []

def change_name_async(poster):
"""Change_name_async will run a background thread
Expand All @@ -182,9 +183,19 @@ def get_name():
def change_name():
embedded_app.DataModel.Project.Name = "foo"

def raise_ex():
raise Exception("Exception")

name.append(poster.post(get_name))
poster.post(change_name)

try:
poster.try_post()
except Exception as e:
error.append(e)

name.append(poster.try_post(get_name))

import threading

change_name_thread = threading.Thread(target=change_name_async, args=(poster,))
Expand All @@ -196,9 +207,11 @@ def change_name():
# thread, e.g. `change_name` that was posted by the poster.
utils.sleep(400)
change_name_thread.join()
assert len(name) == 1
assert len(name) == 2
assert name[0] == "Project"
assert name[1] == "foo"
assert embedded_app.DataModel.Project.Name == "foo"
assert len(error) == 1


@pytest.mark.embedding
Expand Down
Loading