-
Notifications
You must be signed in to change notification settings - Fork 53
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
test Demo #4701
test Demo #4701
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hey @gitofanindya - I've reviewed your changes and found some issues that need to be addressed.
General suggestions:
- Consider abstracting the HTTP request logic into a separate method or service to reduce code duplication and improve maintainability.
- Externalize sensitive information such as URLs and credentials to configuration files or environment variables to enhance security.
- Implement error handling for HTTP responses to gracefully handle cases where expected data might be missing or in an unexpected format.
- Refactor the initialization and setup of new functionalities to maintain a clear separation of concerns and improve code readability.
Here's what I looked at during the review
- 🟡 General issues: 4 issues found
- 🔴 Security: 2 blocking issues, 1 other issue
- 🟢 Testing: all looks good
- 🟡 Complexity: 2 issues found
- 🟢 Docstrings: all looks good
Thanks for using Sourcery. We offer it for free for open source projects and would be very grateful if you could help us grow. If you like it, would you consider sharing Sourcery on your favourite social media? ✨
def fetch_pending_tasks(self, engine: Engine, /) -> None: | ||
log.info("Check for pending approval tasks") | ||
endpoint = "/api/v1/task/" | ||
url = "http://localhost:8080/nuxeo" + endpoint | ||
headers = { | ||
"Accept": "application/json", | ||
"Content-Type": "application/json", | ||
} | ||
try: | ||
# response = NuxeoClient.request(self,method="GET", path=endpoint, headers=headers, ssl_verify=Options.ssl_no_verify) | ||
response = requests.get( | ||
url=url, | ||
verify=True, | ||
timeout=3600, | ||
headers=headers, | ||
auth=("Administrator", "Administrator"), | ||
) | ||
log.info(f">>>>> response type: {type(response)}") | ||
data = response.json() | ||
log.info(f">>>> response: {data}") | ||
if data["resultsCount"] > 0: | ||
log.info("Send pending task notification") | ||
for task in data["entries"]: | ||
log.info(f">>>> task: {task}") | ||
log.info(f">>>> taskid: {task['id']}") | ||
engine.fetch_pending_task_list(task["id"]) | ||
|
||
except Exception: | ||
log.exception("Unable to fetch tasks") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🚨 suggestion (security): Hardcoding the URL and credentials within the fetch_pending_tasks
method could pose a security risk and limit flexibility. Consider externalizing these values to a configuration file or environment variables to enhance security and make the application more adaptable to different environments.
log.info("Check for pending approval tasks") | ||
endpoint = "/api/v1/task/" | ||
url = "http://localhost:8080/nuxeo" + endpoint | ||
headers = { | ||
"Accept": "application/json", | ||
"Content-Type": "application/json", | ||
} | ||
try: | ||
# response = NuxeoClient.request(self,method="GET", path=endpoint, headers=headers, ssl_verify=Options.ssl_no_verify) | ||
response = requests.get( | ||
url=url, | ||
verify=True, | ||
timeout=3600, | ||
headers=headers, | ||
auth=("Administrator", "Administrator"), | ||
) | ||
log.info(f">>>>> response type: {type(response)}") | ||
data = response.json() | ||
log.info(f">>>> response: {data}") | ||
if data["resultsCount"] > 0: | ||
log.info("Send pending task notification") | ||
for task in data["entries"]: | ||
log.info(f">>>> task: {task}") | ||
log.info(f">>>> taskid: {task['id']}") | ||
engine.fetch_pending_task_list(task["id"]) | ||
|
||
except Exception: | ||
log.exception("Unable to fetch tasks") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
issue (code_refinement): The use of a hardcoded URL (http://localhost:8080/nuxeo
) in the fetch_pending_tasks
method might not be suitable for production environments. It's recommended to use a configurable endpoint to ensure the application can be easily adapted to different server configurations without requiring code changes.
response = requests.get( | ||
url=url, | ||
verify=True, | ||
timeout=3600, | ||
headers=headers, | ||
auth=("Administrator", "Administrator"), | ||
) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
suggestion (performance): The fetch_pending_tasks
method uses a very long timeout value (3600 seconds). While this might be necessary in some cases, it could also lead to the application appearing unresponsive if the server takes a long time to respond. Consider allowing this value to be configurable or implementing a more dynamic timeout strategy.
if data["resultsCount"] > 0: | ||
log.info("Send pending task notification") | ||
for task in data["entries"]: | ||
log.info(f">>>> task: {task}") | ||
log.info(f">>>> taskid: {task['id']}") | ||
engine.fetch_pending_task_list(task["id"]) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
suggestion (edge_case_not_handled): The fetch_pending_tasks
method assumes that the response will always contain a resultsCount
and entries
fields. It's a good practice to add error handling for cases where these fields might be missing or not in the expected format to prevent runtime errors.
def get_title(self, uid: str, /) -> str: | ||
endpoint = "/api/v1/task/" | ||
url = "http://localhost:8080/nuxeo" + endpoint | ||
headers = { | ||
"Accept": "application/json", | ||
"Content-Type": "application/json", | ||
} | ||
try: | ||
response = requests.get( | ||
url=url, | ||
verify=True, | ||
timeout=3600, | ||
headers=headers, | ||
auth=("Administrator", "Administrator"), | ||
) | ||
data = response.json() | ||
html = data["entries"][0] | ||
html = html["variables"] | ||
html = html["review_result"] | ||
return html if html else "No Results To Show" | ||
|
||
except Exception: | ||
log.exception("Unable to fetch tasks") | ||
return "No Results Found" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
suggestion (code_refinement): The get_title
method duplicates the logic for making a GET request to the Nuxeo server, which is already present in the fetch_pending_tasks
method. Consider refactoring this into a separate utility function or service to avoid code duplication and simplify maintenance.
html = data["entries"][0] | ||
html = html["variables"] | ||
html = html["review_result"] | ||
return html if html else "No Results To Show" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
suggestion (code-quality): Replace if-expression with or
(or-if-exp-identity
)
return html if html else "No Results To Show" | |
return html or "No Results To Show" |
Explanation
Here we find ourselves setting a value if it evaluates toTrue
, and otherwiseusing a default.
The 'After' case is a bit easier to read and avoids the duplication of
input_currency
.
It works because the left-hand side is evaluated first. If it evaluates to
true then currency
will be set to this and the right-hand side will not be
evaluated. If it evaluates to false the right-hand side will be evaluated and
currency
will be set to DEFAULT_CURRENCY
.
def fetch_pending_tasks(self, engine: Engine, /) -> None: | ||
log.info("Check for pending approval tasks") | ||
endpoint = "/api/v1/task/" | ||
url = "http://localhost:8080/nuxeo" + endpoint |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
suggestion (code-quality): Use f-string instead of string concatenation (use-fstring-for-concatenation
)
url = "http://localhost:8080/nuxeo" + endpoint | |
url = f"http://localhost:8080/nuxeo{endpoint}" |
result = { | ||
k: v for k, v in sorted(result.items(), key=lambda item: item[1]) | ||
} # noqa |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
suggestion (code-quality): Replace identity comprehension with call to collection constructor (identity-comprehension
)
result = { | |
k: v for k, v in sorted(result.items(), key=lambda item: item[1]) | |
} # noqa | |
result = dict(sorted(result.items(), key=lambda item: item[1])) |
Explanation
Convert list/set/tuple comprehensions that do not change the input elements into.Before
# List comprehensions
[item for item in coll]
[item for item in friends.names()]
# Dict comprehensions
{k: v for k, v in coll}
{k: v for k, v in coll.items()} # Only if we know coll is a `dict`
# Unneeded call to `.items()`
dict(coll.items()) # Only if we know coll is a `dict`
# Set comprehensions
{item for item in coll}
After
# List comprehensions
list(iter(coll))
list(iter(friends.names()))
# Dict comprehensions
dict(coll)
dict(coll)
# Unneeded call to `.items()`
dict(coll)
# Set comprehensions
set(coll)
All these comprehensions are just creating a copy of the original collection.
They can all be simplified by simply constructing a new collection directly. The
resulting code is easier to read and shows the intent more clearly.
engine = self._manager.engines.get(uid) | ||
if engine: |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
suggestion (code-quality): Use named expression to simplify assignment and conditional (use-named-expression
)
engine = self._manager.engines.get(uid) | |
if engine: | |
if engine := self._manager.engines.get(uid): |
@pyqtSlot(str, result=str) | ||
def get_title(self, uid: str, /) -> str: | ||
endpoint = "/api/v1/task/" | ||
url = "http://localhost:8080/nuxeo" + endpoint |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
suggestion (code-quality): Use f-string instead of string concatenation (use-fstring-for-concatenation
)
url = "http://localhost:8080/nuxeo" + endpoint | |
url = f"http://localhost:8080/nuxeo{endpoint}" |
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #4701 +/- ##
==========================================
+ Coverage 45.21% 49.04% +3.83%
==========================================
Files 93 94 +1
Lines 15638 15764 +126
==========================================
+ Hits 7071 7732 +661
+ Misses 8567 8032 -535
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. |
No description provided.