Add iucn util to harvest data - #4580
Conversation
There was a problem hiding this comment.
Pull Request Overview
This PR enhances the IUCN API integration by centralizing request logic, improving error handling/pagination support, and adding a management command for manual testing.
- Introduced
_make_iucn_requesthelper to unify API calls and capture pagination headers. - Refactored
get_assessment_detail,fetch_taxa, andget_iucn_statusto use the new helper. - Added
test_iucn_apimanagement command to exercise API methods within a tenant schema.
Reviewed Changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
| bims/utils/iucn.py | New _make_iucn_request, refactored API utility functions, pagination metadata. |
| bims/management/commands/test_iucn_api.py | CLI command to invoke and display IUCN API calls per tenant with options. |
Comments suppressed due to low confidence (2)
bims/utils/iucn.py:11
- The new
_make_iucn_requesthelper and its pagination logic lack unit tests. Consider adding tests for successful calls, error paths, and header parsing to ensure reliability.
def _make_iucn_request(endpoint: str, params: dict = None) -> dict | None:
bims/utils/iucn.py:37
- The except clause references
requests.exceptions.JSONDecodeError, which does not exist. JSON parsing errors raisejson.JSONDecodeErrororValueError. Update the except to catch the correct exception class so parsing failures are handled.
except (HTTPError, SSLError,
| self.stdout.write(self.style.NOTICE(f'Testing get_assessment_detail({assessment_id})...')) | ||
| data = iucn.get_assessment_detail(assessment_id) | ||
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | ||
|
|
||
| if habitat_code: | ||
| self.stdout.write(self.style.NOTICE(f'Testing fetch_taxa("{habitat_code}")...')) | ||
| data = iucn.fetch_taxa(habitat_code) | ||
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | ||
|
|
||
| if taxon_id: | ||
| try: | ||
| taxon = Taxonomy.objects.get(id=taxon_id) | ||
| self.stdout.write(self.style.NOTICE(f'Testing get_iucn_status(taxon={taxon})...')) |
There was a problem hiding this comment.
self.style.NOTICE is not a standard Django BaseCommand style and will raise an AttributeError. Use self.style.WARNING, self.style.ERROR, or define a custom style if NOTICE is required.
| self.stdout.write(self.style.NOTICE(f'Testing get_assessment_detail({assessment_id})...')) | |
| data = iucn.get_assessment_detail(assessment_id) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if habitat_code: | |
| self.stdout.write(self.style.NOTICE(f'Testing fetch_taxa("{habitat_code}")...')) | |
| data = iucn.fetch_taxa(habitat_code) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if taxon_id: | |
| try: | |
| taxon = Taxonomy.objects.get(id=taxon_id) | |
| self.stdout.write(self.style.NOTICE(f'Testing get_iucn_status(taxon={taxon})...')) | |
| self.stdout.write(self.style.WARNING(f'Testing get_assessment_detail({assessment_id})...')) | |
| data = iucn.get_assessment_detail(assessment_id) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if habitat_code: | |
| self.stdout.write(self.style.WARNING(f'Testing fetch_taxa("{habitat_code}")...')) | |
| data = iucn.fetch_taxa(habitat_code) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if taxon_id: | |
| try: | |
| taxon = Taxonomy.objects.get(id=taxon_id) | |
| self.stdout.write(self.style.WARNING(f'Testing get_iucn_status(taxon={taxon})...')) |
| self.stdout.write(self.style.NOTICE(f'Testing get_assessment_detail({assessment_id})...')) | ||
| data = iucn.get_assessment_detail(assessment_id) | ||
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | ||
|
|
||
| if habitat_code: | ||
| self.stdout.write(self.style.NOTICE(f'Testing fetch_taxa("{habitat_code}")...')) | ||
| data = iucn.fetch_taxa(habitat_code) | ||
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | ||
|
|
||
| if taxon_id: | ||
| try: | ||
| taxon = Taxonomy.objects.get(id=taxon_id) | ||
| self.stdout.write(self.style.NOTICE(f'Testing get_iucn_status(taxon={taxon})...')) |
There was a problem hiding this comment.
Same issue here: self.style.NOTICE is undefined in Django's command styles. Switch to a valid style like WARNING or add a NOTICE style.
| self.stdout.write(self.style.NOTICE(f'Testing get_assessment_detail({assessment_id})...')) | |
| data = iucn.get_assessment_detail(assessment_id) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if habitat_code: | |
| self.stdout.write(self.style.NOTICE(f'Testing fetch_taxa("{habitat_code}")...')) | |
| data = iucn.fetch_taxa(habitat_code) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if taxon_id: | |
| try: | |
| taxon = Taxonomy.objects.get(id=taxon_id) | |
| self.stdout.write(self.style.NOTICE(f'Testing get_iucn_status(taxon={taxon})...')) | |
| self.stdout.write(self.style.WARNING(f'Testing get_assessment_detail({assessment_id})...')) | |
| data = iucn.get_assessment_detail(assessment_id) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if habitat_code: | |
| self.stdout.write(self.style.WARNING(f'Testing fetch_taxa("{habitat_code}")...')) | |
| data = iucn.fetch_taxa(habitat_code) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if taxon_id: | |
| try: | |
| taxon = Taxonomy.objects.get(id=taxon_id) | |
| self.stdout.write(self.style.WARNING(f'Testing get_iucn_status(taxon={taxon})...')) |
| self.stdout.write(self.style.NOTICE(f'Testing get_assessment_detail({assessment_id})...')) | ||
| data = iucn.get_assessment_detail(assessment_id) | ||
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | ||
|
|
||
| if habitat_code: | ||
| self.stdout.write(self.style.NOTICE(f'Testing fetch_taxa("{habitat_code}")...')) | ||
| data = iucn.fetch_taxa(habitat_code) | ||
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | ||
|
|
||
| if taxon_id: | ||
| try: | ||
| taxon = Taxonomy.objects.get(id=taxon_id) | ||
| self.stdout.write(self.style.NOTICE(f'Testing get_iucn_status(taxon={taxon})...')) |
There was a problem hiding this comment.
self.style.NOTICE will cause an AttributeError. Consider using a supported style or adding NOTICE to BaseCommand styles.
| self.stdout.write(self.style.NOTICE(f'Testing get_assessment_detail({assessment_id})...')) | |
| data = iucn.get_assessment_detail(assessment_id) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if habitat_code: | |
| self.stdout.write(self.style.NOTICE(f'Testing fetch_taxa("{habitat_code}")...')) | |
| data = iucn.fetch_taxa(habitat_code) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if taxon_id: | |
| try: | |
| taxon = Taxonomy.objects.get(id=taxon_id) | |
| self.stdout.write(self.style.NOTICE(f'Testing get_iucn_status(taxon={taxon})...')) | |
| self.stdout.write(f'Testing get_assessment_detail({assessment_id})...') | |
| data = iucn.get_assessment_detail(assessment_id) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if habitat_code: | |
| self.stdout.write(f'Testing fetch_taxa("{habitat_code}")...') | |
| data = iucn.fetch_taxa(habitat_code) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if taxon_id: | |
| try: | |
| taxon = Taxonomy.objects.get(id=taxon_id) | |
| self.stdout.write(f'Testing get_iucn_status(taxon={taxon})...') |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | ||
|
|
||
| if habitat_code: | ||
| self.stdout.write(self.style.NOTICE(f'Testing fetch_taxa("{habitat_code}")...')) | ||
| data = iucn.fetch_taxa(habitat_code) | ||
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) |
There was a problem hiding this comment.
[nitpick] Wrapping json.dumps in str() is redundant since json.dumps already returns a string. You can pass json.dumps(data, indent=4) directly.
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| if habitat_code: | |
| self.stdout.write(self.style.NOTICE(f'Testing fetch_taxa("{habitat_code}")...')) | |
| data = iucn.fetch_taxa(habitat_code) | |
| self.stdout.write(self.style.SUCCESS(str(json.dumps(data, indent=4)))) | |
| self.stdout.write(self.style.SUCCESS(json.dumps(data, indent=4))) | |
| if habitat_code: | |
| self.stdout.write(self.style.NOTICE(f'Testing fetch_taxa("{habitat_code}")...')) | |
| data = iucn.fetch_taxa(habitat_code) | |
| self.stdout.write(self.style.SUCCESS(json.dumps(data, indent=4))) |
This pull request introduces enhancements to the IUCN API integration, including the addition of a new management command for testing API functions and significant refactoring of the API utility methods. These changes improve code modularity, error handling, and functionality.
New functionality:
bims/management/commands/test_iucn_api.py: Added a new Django management commandtest_iucn_apito test IUCN API functions (get_assessment_detail,fetch_taxa, andget_iucn_status) within a specific tenant schema. This includes argument parsing for tenant schema, assessment ID, habitat code, and taxonomy ID.Refactoring and improvements to API utilities:
bims/utils/iucn.py: Introduced_make_iucn_request, a helper function to centralize IUCN API requests, handle headers, and parse response metadata for pagination. This improves code reuse and error handling.bims/utils/iucn.py: Refactoredget_assessment_detailandfetch_taxato use_make_iucn_request, adding support for structured response data and pagination metadata.bims/utils/iucn.py: Updatedget_iucn_statusto use_make_iucn_request, simplifying the code and removing redundant error handling logic. [1] [2]