Feat/lift encumbrance #824
Open
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
WalkthroughThis change introduces the ability for state admins to lift encumbrances from specific license or privilege records via new PATCH API endpoints. It adds backend logic for transactional updates, new request models, API resource configurations, comprehensive unit and smoke tests, and supporting utilities to ensure proper encumbrance lifting, validation, and record state transitions. Changes
Sequence Diagram(s)sequenceDiagram
participant Admin as State Admin
participant API as API Gateway
participant Handler as Lambda Handler
participant DataClient as DataClient
participant DynamoDB as DynamoDB
Admin->>API: PATCH /license/encumbrance or /privilege/encumbrance
API->>Handler: Invoke handler with event
Handler->>Handler: Validate input, parse lift date & encumbrance ID
Handler->>DataClient: lift_license_encumbrance(...) or lift_privilege_encumbrance(...)
DataClient->>DynamoDB: Retrieve provider user records
DataClient->>DynamoDB: Transactional update (adverse action, license/privilege, provider)
DynamoDB-->>DataClient: Transaction result
DataClient-->>Handler: Success or error
Handler-->>API: 200 OK or error response
API-->>Admin: Response
Assessment against linked issues
Possibly related PRs
Suggested reviewers
Poem
✨ Finishing Touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
If all encumbrances have been lifted for this provider, we need to set their encumbrance status to unencumbered
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (5)
backend/compact-connect/lambdas/python/common/cc_common/data_model/provider_record_util.py (1)
368-398: Consider using date objects directly for sorting.While sorting by
isoformat()works correctly for ISO dates, using the date objects directly would be more robust and avoid string conversion overhead.- key=lambda x: x.dateOfIssuance.isoformat(), + key=lambda x: x.dateOfIssuance,This change applies to all three sorting operations in this method.
backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_encumbrance.py (1)
781-781: Use consistent datetime parameter syntax.For consistency with line 477, use the positional argument form.
- future_date = (datetime.now(tz=UTC) + timedelta(days=2)).strftime('%Y-%m-%d') + future_date = (datetime.now(UTC) + timedelta(days=2)).strftime('%Y-%m-%d')backend/compact-connect/lambdas/python/common/cc_common/data_model/data_client.py (3)
165-198: Consider using the existing pagination pattern for consistency.The method correctly implements pagination, but could be improved:
- Consider using the
@paginated_querydecorator pattern used elsewhere in this class for consistency- Add exception handling for DynamoDB operations
Here's a more consistent approach using the existing pagination pattern:
- @logger_inject_kwargs(logger, 'compact', 'provider_id') - def get_provider_user_records( - self, - *, - compact: str, - provider_id: str, - consistent_read: bool = True, - ) -> ProviderUserRecords: - logger.info('Getting provider') - - resp = {'Items': []} - last_evaluated_key = None - - while True: - pagination = {'ExclusiveStartKey': last_evaluated_key} if last_evaluated_key else {} - - query_resp = self.config.provider_table.query( - Select='ALL_ATTRIBUTES', - KeyConditionExpression=Key('pk').eq(f'{compact}#PROVIDER#{provider_id}') - & Key('sk').begins_with(f'{compact}#PROVIDER'), - ConsistentRead=consistent_read, - **pagination, - ) - - resp['Items'].extend(query_resp.get('Items', [])) - - last_evaluated_key = query_resp.get('LastEvaluatedKey') - if not last_evaluated_key: - break - if not resp['Items']: - raise CCNotFoundException('Provider not found') - - return ProviderUserRecords(resp['Items']) + @logger_inject_kwargs(logger, 'compact', 'provider_id') + def get_provider_user_records( + self, + *, + compact: str, + provider_id: str, + consistent_read: bool = True, + ) -> ProviderUserRecords: + logger.info('Getting provider') + + all_items = [] + try: + # Use the existing get_provider method with detail=True to get all records + result = self.get_provider( + compact=compact, + provider_id=provider_id, + detail=True, + consistent_read=consistent_read + ) + all_items.extend(result.get('Items', [])) + + # Handle any remaining pages + while 'LastEvaluatedKey' in result: + result = self.get_provider( + compact=compact, + provider_id=provider_id, + detail=True, + consistent_read=consistent_read, + dynamo_pagination={'ExclusiveStartKey': result['LastEvaluatedKey']} + ) + all_items.extend(result.get('Items', [])) + + except CCNotFoundException: + raise + except ClientError as e: + logger.error('Failed to retrieve provider records', error=str(e)) + raise CCAwsServiceException('Failed to retrieve provider records') from e + + return ProviderUserRecords(all_items)
1100-1166: Refactor to reduce code duplication and improve type checking.The logic is correct but can be improved:
- Using
getattrfor type checking (line 1121) is fragile- The filtering logic is duplicated between licenses and privileges
Consider this refactored approach:
def _generate_provider_encumbered_status_transaction_items_if_no_encumbrances( self, provider_user_records: ProviderUserRecords, lifted_record: PrivilegeData | LicenseData ) -> list[dict]: """ Check if any licenses or privileges (excluding the lifted record) still have encumbered status. If none are encumbered, return transaction items to set the provider record to unencumbered. :param ProviderUserRecords provider_user_records: All provider records :param lifted_record: The privilege or license record that is having its encumbrance lifted :return: List of transaction items (empty if other records are still encumbered) """ # Get the provider record provider_record = provider_user_records.get_provider_record() # Get all license records license_records = provider_user_records.get_license_records() # Get all privilege records privilege_records = provider_user_records.get_privilege_records() - # Check if the lifted record is a license or privilege based on its type - lifted_record_type = getattr(lifted_record, 'type', None) + # Determine if lifted record is a license or privilege + is_lifted_license = isinstance(lifted_record, LicenseData) + + def is_same_record(record, lifted): + """Check if a record matches the lifted record.""" + return (record.jurisdiction == lifted.jurisdiction and + record.licenseType == lifted.licenseType) # Check license records for encumbered status (excluding the lifted record if it's a license) for license_record in license_records: - if ( - lifted_record_type == 'license' - and license_record.jurisdiction == lifted_record.jurisdiction - and license_record.licenseType == lifted_record.licenseType - ): + if is_lifted_license and is_same_record(license_record, lifted_record): # Skip the record being lifted continue if license_record.encumberedStatus == LicenseEncumberedStatusEnum.ENCUMBERED: logger.info( 'License record still encumbered, provider record will not be updated', encumbered_license_jurisdiction=license_record.jurisdiction, encumbered_license_type=license_record.licenseType, ) return [] # Check privilege records for encumbered status (excluding the lifted record if it's a privilege) for privilege_record in privilege_records: - if ( - lifted_record_type == 'privilege' - and privilege_record.jurisdiction == lifted_record.jurisdiction - and privilege_record.licenseType == lifted_record.licenseType - ): + if not is_lifted_license and is_same_record(privilege_record, lifted_record): # Skip the record being lifted continue if privilege_record.encumberedStatus == PrivilegeEncumberedStatusEnum.ENCUMBERED: logger.info( 'Privilege record still encumbered, provider record will not be updated', encumbered_privilege_jurisdiction=privilege_record.jurisdiction, encumbered_privilege_type=privilege_record.licenseType, ) return []
1491-1624: Consider refactoring to reduce duplication with lift_privilege_encumbrance.This method is nearly identical to
lift_privilege_encumbrance. The same error handling issue exists at line 1563.Consider extracting the common logic into a shared private method:
def _lift_encumbrance( self, compact: str, provider_id: str, jurisdiction: str, license_type_abbreviation: str, adverse_action_id: str, effective_lift_date: date, lifting_user: str, record_type: Literal['license', 'privilege'], ) -> None: """ Generic method to lift an encumbrance from either a license or privilege record. """ # Common implementation here...This would eliminate the duplication and make maintenance easier.
Also apply the same error handling fix for the license type lookup as suggested for
lift_privilege_encumbrance.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (12)
backend/compact-connect/lambdas/python/common/cc_common/data_model/data_client.py(6 hunks)backend/compact-connect/lambdas/python/common/cc_common/data_model/provider_record_util.py(2 hunks)backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/common.py(1 hunks)backend/compact-connect/lambdas/python/provider-data-v1/handlers/encumbrance.py(4 hunks)backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_encumbrance.py(2 hunks)backend/compact-connect/stacks/api_stack/v1_api/api_model.py(1 hunks)backend/compact-connect/stacks/api_stack/v1_api/provider_management.py(2 hunks)backend/compact-connect/tests/app/test_api/test_provider_management_api.py(5 hunks)backend/compact-connect/tests/resources/snapshots/LICENSE_ENCUMBRANCE_LIFTING_REQUEST_SCHEMA.json(1 hunks)backend/compact-connect/tests/resources/snapshots/PRIVILEGE_ENCUMBRANCE_LIFTING_REQUEST_SCHEMA.json(1 hunks)backend/compact-connect/tests/smoke/encumbrance_smoke_tests.py(1 hunks)backend/compact-connect/tests/smoke/smoke_common.py(2 hunks)
🧰 Additional context used
🧬 Code Graph Analysis (2)
backend/compact-connect/tests/smoke/smoke_common.py (1)
backend/compact-connect/tests/smoke/config.py (1)
provider_user_dynamodb_table(20-21)
backend/compact-connect/tests/smoke/encumbrance_smoke_tests.py (3)
backend/compact-connect/tests/smoke/purchasing_privileges_smoke_tests.py (1)
test_purchasing_privilege(106-239)backend/compact-connect/tests/smoke/smoke_common.py (9)
SmokeTestFailureException(12-18)call_provider_users_me_endpoint(218-226)create_test_staff_user(100-122)delete_test_staff_user(82-97)get_all_provider_database_records(229-239)get_license_type_abbreviation(174-181)get_provider_user_dynamodb_table(188-189)get_staff_user_auth_headers(167-171)load_smoke_test_env(212-215)backend/compact-connect/tests/smoke/config.py (1)
api_base_url(16-17)
🔇 Additional comments (15)
backend/compact-connect/lambdas/python/common/cc_common/data_model/schema/common.py (1)
226-226: LGTM! Clean enum addition for lifting encumbrance functionality.The new
LIFTING_ENCUMBRANCEenum member follows the established naming convention and appropriately represents the update category for lifting encumbrances.backend/compact-connect/tests/smoke/smoke_common.py (2)
7-7: LGTM! Necessary import for DynamoDB query condition.The
Keyimport is required for the newget_all_provider_database_records()function.
229-239: LGTM! Well-implemented helper function for testing.The
get_all_provider_database_records()function provides a clean utility for retrieving all provider database records during testing. The implementation correctly:
- Uses the existing
call_provider_users_me_endpoint()to get provider context- Constructs the appropriate partition key format
{compact}#PROVIDER#{provider_id}- Performs a DynamoDB query with proper key condition
- Returns the query results
This will be valuable for validating encumbrance lifting functionality in smoke tests.
backend/compact-connect/stacks/api_stack/v1_api/provider_management.py (2)
634-650:✅ Verification successful
LGTM! Well-structured PATCH endpoint for lifting privilege encumbrances.
The implementation correctly follows the established pattern of the existing POST method with appropriate authorization, validation, and response handling. Using the same Lambda handler (
provider_encumbrance_handler) for both POST and PATCH operations is a good approach for code reuse.Let me verify that the referenced request model exists and the Lambda handler properly routes PATCH requests:
🏁 Script executed:
#!/bin/bash # Description: Verify the patch privilege encumbrance request model exists and Lambda handler supports PATCH routing # Test: Search for the patch_privilege_encumbrance_request_model definition echo "=== Searching for patch_privilege_encumbrance_request_model ===" rg -A 5 "patch_privilege_encumbrance_request_model" # Test: Search for PATCH method handling in the encumbrance handler echo "=== Searching for PATCH method handling in encumbrance handler ===" rg -A 10 -B 5 "PATCH|patch" --type py | grep -A 10 -B 5 "encumbrance"Length of output: 59012
Verification complete – PATCH encumbrance endpoints are correctly wired
- api_model.py defines
patch_privilege_encumbrance_request_modelandpatch_license_encumbrance_request_modelwith the requiredeffectiveLiftDateandencumbranceIdfields.- handlers/encumbrance.py routes
httpMethod == 'PATCH'tohandle_privilege_encumbrance_liftingandhandle_license_encumbrance_lifting.- Smoke and unit tests exercise both PATCH endpoints successfully.
No further changes needed.
676-692:✅ Verification successful
LGTM! Consistent implementation for lifting license encumbrances.
The license encumbrance PATCH method implementation is consistent with the privilege encumbrance PATCH method, maintaining the same security, validation, and integration patterns. This consistency ensures predictable behavior across both encumbrance types.
Let me verify the corresponding request model for license encumbrances:
🏁 Script executed:
#!/bin/bash # Description: Verify the patch license encumbrance request model exists # Test: Search for the patch_license_encumbrance_request_model definition echo "=== Searching for patch_license_encumbrance_request_model ===" rg -A 5 "patch_license_encumbrance_request_model" # Test: Verify consistency between privilege and license patch models echo "=== Comparing privilege and license patch models ===" rg -A 10 "patch_.*_encumbrance_request_model"Length of output: 11688
LGTM – PATCH license encumbrance request model verified
Verified that thepatch_license_encumbrance_request_modelis defined inapi_model.pyand matches thepatch_privilege_encumbrance_request_modelschema (same required properties and structure). The PATCH method inprovider_management.pyis consistent with the privilege encumbrance implementation, preserving identical security, validation, and integration patterns.backend/compact-connect/stacks/api_stack/v1_api/api_model.py (1)
443-503: LGTM! Well-structured request models for encumbrance lifting.The implementation correctly follows the established patterns in the codebase with proper caching, validation schemas, and clear documentation. Both models appropriately validate the date format and enforce length constraints on the encumbrance ID.
backend/compact-connect/lambdas/python/provider-data-v1/handlers/encumbrance.py (1)
65-65: Good improvement to error messaging.Including the detailed validation errors helps developers debug request issues more effectively.
backend/compact-connect/lambdas/python/common/cc_common/data_model/provider_record_util.py (1)
247-416: Well-designed utility class for provider record management.The
ProviderUserRecordsclass provides a clean abstraction for working with provider records, with appropriate separation of concerns and clear method documentation. The implementation follows good practices for data encapsulation and type safety.backend/compact-connect/lambdas/python/provider-data-v1/tests/function/test_handlers/test_encumbrance.py (1)
424-1031: Excellent test coverage for the new encumbrance lifting functionality!The test classes
TestPatchPrivilegeEncumbranceLiftingandTestPatchLicenseEncumbranceLiftingprovide comprehensive coverage including:
- Input validation (future dates, invalid date formats)
- Error scenarios (not found, already lifted)
- Successful lifting operations with proper state transitions
- Authorization enforcement (only state admins allowed)
- Edge cases (multiple encumbrances, mixed license/privilege encumbrances)
The use of helper methods reduces duplication and improves maintainability.
backend/compact-connect/tests/smoke/encumbrance_smoke_tests.py (1)
1-566: Well-structured end-to-end smoke tests for encumbrance functionality!The smoke tests provide excellent coverage of the complete encumbrance lifecycle:
- Proper test environment setup and cleanup
- Comprehensive workflow testing (create → verify → lift → verify)
- Good error handling with descriptive failure messages
- Proper test isolation with user creation/deletion
The helper functions and clear logging make the tests maintainable and debuggable.
backend/compact-connect/tests/app/test_api/test_provider_management_api.py (2)
23-179: Excellent refactoring with helper methods!The introduction of helper methods
_get_privilege_encumbrance_resource_id,_get_privilege_license_type_param_resource_id, and_get_license_encumbrance_resource_idsignificantly improves code maintainability by:
- Eliminating duplicate resource traversal logic
- Making the tests more readable
- Providing a single source of truth for resource hierarchy navigation
690-794: Comprehensive test coverage for the new PATCH endpoints!The new tests
test_synth_generates_privilege_encumbrance_lifting_endpointandtest_synth_generates_license_encumbrance_lifting_endpointproperly verify:
- Correct HTTP method (PATCH)
- Proper authorization configuration
- Request model validation
- Integration with the encumbrance handler
- Snapshot testing for schema validation
These tests follow the established patterns and ensure the PATCH endpoints are configured correctly in the API Gateway.
backend/compact-connect/lambdas/python/common/cc_common/data_model/data_client.py (3)
12-12: LGTM!The import is properly placed and necessary for the new
ProviderUserRecordsfunctionality.
1017-1019: LGTM!Clean helper method for generating Put transaction items.
1020-1049: LGTM!Well-implemented method that correctly generates the Update transaction item for lifting an adverse action.
| "effectiveLiftDate": { | ||
| "description": "The effective date when the encumbrance will be lifted", | ||
| "format": "date", | ||
| "pattern": "^[12]{1}[0-9]{3}-[01]{1}[0-9]{1}-[0-3]{1}[0-9]{1}$", |
There was a problem hiding this comment.
Fix the date validation pattern to prevent invalid dates.
The current regex pattern ^[12]{1}[0-9]{3}-[01]{1}[0-9]{1}-[0-3]{1}[0-9]{1}$ incorrectly allows invalid dates:
- Month part
[01]{1}[0-9]{1}accepts 00-19, but valid months are 01-12 - Day part
[0-3]{1}[0-9]{1}accepts 00-39, but valid days are 01-31 and 00 is invalid
Apply this fix to use a proper date validation pattern:
- "pattern": "^[12]{1}[0-9]{3}-[01]{1}[0-9]{1}-[0-3]{1}[0-9]{1}$",
+ "pattern": "^[12]{1}[0-9]{3}-(0[1-9]|1[0-2])-(0[1-9]|[12][0-9]|3[01])$",This pattern correctly validates:
- Month: 01-12 using
(0[1-9]|1[0-2]) - Day: 01-31 using
(0[1-9]|[12][0-9]|3[01])
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| "pattern": "^[12]{1}[0-9]{3}-[01]{1}[0-9]{1}-[0-3]{1}[0-9]{1}$", | |
| "pattern": "^[12]{1}[0-9]{3}-(0[1-9]|1[0-2])-(0[1-9]|[12][0-9]|3[01])$", |
🤖 Prompt for AI Agents
In
backend/compact-connect/tests/resources/snapshots/PRIVILEGE_ENCUMBRANCE_LIFTING_REQUEST_SCHEMA.json
at line 7, the date regex pattern allows invalid months and days. Replace the
month part `[01]{1}[0-9]{1}` with `(0[1-9]|1[0-2])` to restrict months to 01-12,
and replace the day part `[0-3]{1}[0-9]{1}` with `(0[1-9]|[12][0-9]|3[01])` to
restrict days to 01-31, ensuring the pattern only matches valid dates.
| "effectiveLiftDate": { | ||
| "description": "The effective date when the encumbrance will be lifted", | ||
| "format": "date", | ||
| "pattern": "^[12]{1}[0-9]{3}-[01]{1}[0-9]{1}-[0-3]{1}[0-9]{1}$", |
There was a problem hiding this comment.
Fix the date validation pattern to prevent invalid dates.
The current regex pattern ^[12]{1}[0-9]{3}-[01]{1}[0-9]{1}-[0-3]{1}[0-9]{1}$ incorrectly allows invalid dates:
- Month part
[01]{1}[0-9]{1}accepts 00-19, but valid months are 01-12 - Day part
[0-3]{1}[0-9]{1}accepts 00-39, but valid days are 01-31 and 00 is invalid
Apply this fix to use a proper date validation pattern:
- "pattern": "^[12]{1}[0-9]{3}-[01]{1}[0-9]{1}-[0-3]{1}[0-9]{1}$",
+ "pattern": "^[12]{1}[0-9]{3}-(0[1-9]|1[0-2])-(0[1-9]|[12][0-9]|3[01])$",This pattern correctly validates:
- Month: 01-12 using
(0[1-9]|1[0-2]) - Day: 01-31 using
(0[1-9]|[12][0-9]|3[01])
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| "pattern": "^[12]{1}[0-9]{3}-[01]{1}[0-9]{1}-[0-3]{1}[0-9]{1}$", | |
| "pattern": "^[12]{1}[0-9]{3}-(0[1-9]|1[0-2])-(0[1-9]|[12][0-9]|3[01])$", |
🤖 Prompt for AI Agents
In
backend/compact-connect/tests/resources/snapshots/LICENSE_ENCUMBRANCE_LIFTING_REQUEST_SCHEMA.json
at line 7, the date validation regex pattern allows invalid months and days.
Replace the current pattern "^[12]{1}[0-9]{3}-[01]{1}[0-9]{1}-[0-3]{1}[0-9]{1}$"
with a stricter pattern that validates months as 01-12 using "(0[1-9]|1[0-2])"
and days as 01-31 using "(0[1-9]|[12][0-9]|3[01])" to prevent invalid dates.
Comment on lines
+116
to
+201
| def handle_privilege_encumbrance_lifting(event: dict) -> dict: | ||
| """Handle lifting encumbrance from a privilege record""" | ||
| # Get the cognito sub of the caller for tracing | ||
| cognito_sub = _get_submitting_user_id(event) | ||
|
|
||
| with logger.append_context_keys(cognito_sub=cognito_sub): | ||
| logger.info('Processing privilege encumbrance lifting') | ||
|
|
||
| # Extract path parameters | ||
| compact = event['pathParameters']['compact'] | ||
| provider_id = event['pathParameters']['providerId'] | ||
| jurisdiction = event['pathParameters']['jurisdiction'] | ||
| license_type_abbreviation = event['pathParameters']['licenseType'].lower() | ||
|
|
||
| # Parse and validate request body | ||
| body = json.loads(event['body']) | ||
| effective_lift_date = body['effectiveLiftDate'] | ||
| encumbrance_id = body['encumbranceId'] | ||
|
|
||
| # Validate date format and parse the effective lift date | ||
| try: | ||
| lift_date = datetime.fromisoformat(effective_lift_date).date() | ||
| except ValueError as e: | ||
| raise CCInvalidRequestException('Invalid date format. Expected ISO format (YYYY-MM-DD)') from e | ||
|
|
||
| current_date = config.expiration_resolution_date | ||
|
|
||
| if lift_date > current_date: | ||
| raise CCInvalidRequestException('The lift date must not be a future date') | ||
|
|
||
| # Call the data client method to lift the privilege encumbrance | ||
| config.data_client.lift_privilege_encumbrance( | ||
| compact=compact, | ||
| provider_id=provider_id, | ||
| jurisdiction=jurisdiction, | ||
| license_type_abbreviation=license_type_abbreviation, | ||
| adverse_action_id=encumbrance_id, | ||
| effective_lift_date=lift_date, | ||
| lifting_user=cognito_sub, | ||
| ) | ||
|
|
||
| return {'message': 'OK'} | ||
|
|
||
|
|
||
| def handle_license_encumbrance_lifting(event: dict) -> dict: | ||
| """Handle lifting encumbrance from a license record""" | ||
| # Get the cognito sub of the caller for tracing | ||
| cognito_sub = _get_submitting_user_id(event) | ||
|
|
||
| with logger.append_context_keys(cognito_sub=cognito_sub): | ||
| logger.info('Processing license encumbrance lifting') | ||
|
|
||
| # Extract path parameters | ||
| compact = event['pathParameters']['compact'] | ||
| provider_id = event['pathParameters']['providerId'] | ||
| jurisdiction = event['pathParameters']['jurisdiction'] | ||
| license_type_abbreviation = event['pathParameters']['licenseType'].lower() | ||
|
|
||
| # Parse and validate request body | ||
| body = json.loads(event['body']) | ||
| effective_lift_date = body['effectiveLiftDate'] | ||
| encumbrance_id = body['encumbranceId'] | ||
|
|
||
| # Validate date format and parse the effective lift date | ||
| try: | ||
| lift_date = datetime.fromisoformat(effective_lift_date).date() | ||
| except ValueError as e: | ||
| raise CCInvalidRequestException('Invalid date format. Expected ISO format (YYYY-MM-DD)') from e | ||
|
|
||
| current_date = config.expiration_resolution_date | ||
|
|
||
| if lift_date > current_date: | ||
| raise CCInvalidRequestException('The lift date must not be a future date') | ||
|
|
||
| # Call the data client method to lift the license encumbrance | ||
| config.data_client.lift_license_encumbrance( | ||
| compact=compact, | ||
| provider_id=provider_id, | ||
| jurisdiction=jurisdiction, | ||
| license_type_abbreviation=license_type_abbreviation, | ||
| adverse_action_id=encumbrance_id, | ||
| effective_lift_date=lift_date, | ||
| lifting_user=cognito_sub, | ||
| ) | ||
|
|
||
| return {'message': 'OK'} |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Consider refactoring to reduce code duplication.
The handle_privilege_encumbrance_lifting and handle_license_encumbrance_lifting functions share nearly identical logic. Only the final data client method call differs.
Consider extracting the common logic into a shared function:
+def _validate_and_parse_lift_request(event: dict) -> tuple[str, str, str, str, date, str, str]:
+ """Extract and validate common parameters for lifting encumbrances."""
+ # Get the cognito sub of the caller for tracing
+ cognito_sub = _get_submitting_user_id(event)
+
+ # Extract path parameters
+ compact = event['pathParameters']['compact']
+ provider_id = event['pathParameters']['providerId']
+ jurisdiction = event['pathParameters']['jurisdiction']
+ license_type_abbreviation = event['pathParameters']['licenseType'].lower()
+
+ # Parse and validate request body
+ body = json.loads(event['body'])
+ effective_lift_date = body['effectiveLiftDate']
+ encumbrance_id = body['encumbranceId']
+
+ # Validate date format and parse the effective lift date
+ try:
+ lift_date = datetime.fromisoformat(effective_lift_date).date()
+ except ValueError as e:
+ raise CCInvalidRequestException('Invalid date format. Expected ISO format (YYYY-MM-DD)') from e
+
+ current_date = config.expiration_resolution_date
+
+ if lift_date > current_date:
+ raise CCInvalidRequestException('The lift date must not be a future date')
+
+ return compact, provider_id, jurisdiction, license_type_abbreviation, lift_date, encumbrance_id, cognito_sub
def handle_privilege_encumbrance_lifting(event: dict) -> dict:
"""Handle lifting encumbrance from a privilege record"""
- # Get the cognito sub of the caller for tracing
- cognito_sub = _get_submitting_user_id(event)
-
- with logger.append_context_keys(cognito_sub=cognito_sub):
+ compact, provider_id, jurisdiction, license_type_abbreviation, lift_date, encumbrance_id, cognito_sub = (
+ _validate_and_parse_lift_request(event)
+ )
+
+ with logger.append_context_keys(cognito_sub=cognito_sub):
logger.info('Processing privilege encumbrance lifting')
-
- # Extract path parameters
- compact = event['pathParameters']['compact']
- provider_id = event['pathParameters']['providerId']
- jurisdiction = event['pathParameters']['jurisdiction']
- license_type_abbreviation = event['pathParameters']['licenseType'].lower()
-
- # Parse and validate request body
- body = json.loads(event['body'])
- effective_lift_date = body['effectiveLiftDate']
- encumbrance_id = body['encumbranceId']
-
- # Validate date format and parse the effective lift date
- try:
- lift_date = datetime.fromisoformat(effective_lift_date).date()
- except ValueError as e:
- raise CCInvalidRequestException('Invalid date format. Expected ISO format (YYYY-MM-DD)') from e
-
- current_date = config.expiration_resolution_date
-
- if lift_date > current_date:
- raise CCInvalidRequestException('The lift date must not be a future date')
-
+
# Call the data client method to lift the privilege encumbrance
config.data_client.lift_privilege_encumbrance(
compact=compact,
provider_id=provider_id,
jurisdiction=jurisdiction,
license_type_abbreviation=license_type_abbreviation,
adverse_action_id=encumbrance_id,
effective_lift_date=lift_date,
lifting_user=cognito_sub,
)
-
+
return {'message': 'OK'}
def handle_license_encumbrance_lifting(event: dict) -> dict:
"""Handle lifting encumbrance from a license record"""
- # Get the cognito sub of the caller for tracing
- cognito_sub = _get_submitting_user_id(event)
-
- with logger.append_context_keys(cognito_sub=cognito_sub):
+ compact, provider_id, jurisdiction, license_type_abbreviation, lift_date, encumbrance_id, cognito_sub = (
+ _validate_and_parse_lift_request(event)
+ )
+
+ with logger.append_context_keys(cognito_sub=cognito_sub):
logger.info('Processing license encumbrance lifting')
-
- # Extract path parameters
- compact = event['pathParameters']['compact']
- provider_id = event['pathParameters']['providerId']
- jurisdiction = event['pathParameters']['jurisdiction']
- license_type_abbreviation = event['pathParameters']['licenseType'].lower()
-
- # Parse and validate request body
- body = json.loads(event['body'])
- effective_lift_date = body['effectiveLiftDate']
- encumbrance_id = body['encumbranceId']
-
- # Validate date format and parse the effective lift date
- try:
- lift_date = datetime.fromisoformat(effective_lift_date).date()
- except ValueError as e:
- raise CCInvalidRequestException('Invalid date format. Expected ISO format (YYYY-MM-DD)') from e
-
- current_date = config.expiration_resolution_date
-
- if lift_date > current_date:
- raise CCInvalidRequestException('The lift date must not be a future date')
-
+
# Call the data client method to lift the license encumbrance
config.data_client.lift_license_encumbrance(
compact=compact,
provider_id=provider_id,
jurisdiction=jurisdiction,
license_type_abbreviation=license_type_abbreviation,
adverse_action_id=encumbrance_id,
effective_lift_date=lift_date,
lifting_user=cognito_sub,
)
-
+
return {'message': 'OK'}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def handle_privilege_encumbrance_lifting(event: dict) -> dict: | |
| """Handle lifting encumbrance from a privilege record""" | |
| # Get the cognito sub of the caller for tracing | |
| cognito_sub = _get_submitting_user_id(event) | |
| with logger.append_context_keys(cognito_sub=cognito_sub): | |
| logger.info('Processing privilege encumbrance lifting') | |
| # Extract path parameters | |
| compact = event['pathParameters']['compact'] | |
| provider_id = event['pathParameters']['providerId'] | |
| jurisdiction = event['pathParameters']['jurisdiction'] | |
| license_type_abbreviation = event['pathParameters']['licenseType'].lower() | |
| # Parse and validate request body | |
| body = json.loads(event['body']) | |
| effective_lift_date = body['effectiveLiftDate'] | |
| encumbrance_id = body['encumbranceId'] | |
| # Validate date format and parse the effective lift date | |
| try: | |
| lift_date = datetime.fromisoformat(effective_lift_date).date() | |
| except ValueError as e: | |
| raise CCInvalidRequestException('Invalid date format. Expected ISO format (YYYY-MM-DD)') from e | |
| current_date = config.expiration_resolution_date | |
| if lift_date > current_date: | |
| raise CCInvalidRequestException('The lift date must not be a future date') | |
| # Call the data client method to lift the privilege encumbrance | |
| config.data_client.lift_privilege_encumbrance( | |
| compact=compact, | |
| provider_id=provider_id, | |
| jurisdiction=jurisdiction, | |
| license_type_abbreviation=license_type_abbreviation, | |
| adverse_action_id=encumbrance_id, | |
| effective_lift_date=lift_date, | |
| lifting_user=cognito_sub, | |
| ) | |
| return {'message': 'OK'} | |
| def handle_license_encumbrance_lifting(event: dict) -> dict: | |
| """Handle lifting encumbrance from a license record""" | |
| # Get the cognito sub of the caller for tracing | |
| cognito_sub = _get_submitting_user_id(event) | |
| with logger.append_context_keys(cognito_sub=cognito_sub): | |
| logger.info('Processing license encumbrance lifting') | |
| # Extract path parameters | |
| compact = event['pathParameters']['compact'] | |
| provider_id = event['pathParameters']['providerId'] | |
| jurisdiction = event['pathParameters']['jurisdiction'] | |
| license_type_abbreviation = event['pathParameters']['licenseType'].lower() | |
| # Parse and validate request body | |
| body = json.loads(event['body']) | |
| effective_lift_date = body['effectiveLiftDate'] | |
| encumbrance_id = body['encumbranceId'] | |
| # Validate date format and parse the effective lift date | |
| try: | |
| lift_date = datetime.fromisoformat(effective_lift_date).date() | |
| except ValueError as e: | |
| raise CCInvalidRequestException('Invalid date format. Expected ISO format (YYYY-MM-DD)') from e | |
| current_date = config.expiration_resolution_date | |
| if lift_date > current_date: | |
| raise CCInvalidRequestException('The lift date must not be a future date') | |
| # Call the data client method to lift the license encumbrance | |
| config.data_client.lift_license_encumbrance( | |
| compact=compact, | |
| provider_id=provider_id, | |
| jurisdiction=jurisdiction, | |
| license_type_abbreviation=license_type_abbreviation, | |
| adverse_action_id=encumbrance_id, | |
| effective_lift_date=lift_date, | |
| lifting_user=cognito_sub, | |
| ) | |
| return {'message': 'OK'} | |
| def _validate_and_parse_lift_request(event: dict) -> tuple[str, str, str, str, date, str, str]: | |
| """Extract and validate common parameters for lifting encumbrances.""" | |
| # Get the cognito sub of the caller for tracing | |
| cognito_sub = _get_submitting_user_id(event) | |
| # Extract path parameters | |
| compact = event['pathParameters']['compact'] | |
| provider_id = event['pathParameters']['providerId'] | |
| jurisdiction = event['pathParameters']['jurisdiction'] | |
| license_type_abbreviation = event['pathParameters']['licenseType'].lower() | |
| # Parse and validate request body | |
| body = json.loads(event['body']) | |
| effective_lift_date = body['effectiveLiftDate'] | |
| encumbrance_id = body['encumbranceId'] | |
| # Validate date format and parse the effective lift date | |
| try: | |
| lift_date = datetime.fromisoformat(effective_lift_date).date() | |
| except ValueError as e: | |
| raise CCInvalidRequestException('Invalid date format. Expected ISO format (YYYY-MM-DD)') from e | |
| current_date = config.expiration_resolution_date | |
| if lift_date > current_date: | |
| raise CCInvalidRequestException('The lift date must not be a future date') | |
| return compact, provider_id, jurisdiction, license_type_abbreviation, lift_date, encumbrance_id, cognito_sub | |
| def handle_privilege_encumbrance_lifting(event: dict) -> dict: | |
| """Handle lifting encumbrance from a privilege record""" | |
| compact, provider_id, jurisdiction, license_type_abbreviation, lift_date, encumbrance_id, cognito_sub = ( | |
| _validate_and_parse_lift_request(event) | |
| ) | |
| with logger.append_context_keys(cognito_sub=cognito_sub): | |
| logger.info('Processing privilege encumbrance lifting') | |
| # Call the data client method to lift the privilege encumbrance | |
| config.data_client.lift_privilege_encumbrance( | |
| compact=compact, | |
| provider_id=provider_id, | |
| jurisdiction=jurisdiction, | |
| license_type_abbreviation=license_type_abbreviation, | |
| adverse_action_id=encumbrance_id, | |
| effective_lift_date=lift_date, | |
| lifting_user=cognito_sub, | |
| ) | |
| return {'message': 'OK'} | |
| def handle_license_encumbrance_lifting(event: dict) -> dict: | |
| """Handle lifting encumbrance from a license record""" | |
| compact, provider_id, jurisdiction, license_type_abbreviation, lift_date, encumbrance_id, cognito_sub = ( | |
| _validate_and_parse_lift_request(event) | |
| ) | |
| with logger.append_context_keys(cognito_sub=cognito_sub): | |
| logger.info('Processing license encumbrance lifting') | |
| # Call the data client method to lift the license encumbrance | |
| config.data_client.lift_license_encumbrance( | |
| compact=compact, | |
| provider_id=provider_id, | |
| jurisdiction=jurisdiction, | |
| license_type_abbreviation=license_type_abbreviation, | |
| adverse_action_id=encumbrance_id, | |
| effective_lift_date=lift_date, | |
| lifting_user=cognito_sub, | |
| ) | |
| return {'message': 'OK'} |
🤖 Prompt for AI Agents
In
backend/compact-connect/lambdas/python/provider-data-v1/handlers/encumbrance.py
between lines 116 and 201, the functions handle_privilege_encumbrance_lifting
and handle_license_encumbrance_lifting contain nearly identical code except for
the final data client method called. Refactor by extracting the shared
logic—such as extracting path parameters, parsing and validating the request
body, and date validation—into a single helper function that accepts the
specific data client method as a parameter. Then, update the original functions
to call this helper with the appropriate method, reducing duplication and
improving maintainability.
Comment on lines
+258
to
+283
| def get_privilege_records( | ||
| self, | ||
| filter_condition: Callable[[PrivilegeData], bool] | None = None, | ||
| ) -> list[PrivilegeData]: | ||
| """ | ||
| Get all privilege records from a list of provider records. | ||
| :param filter_condition: An optional filter to apply to the privilege records | ||
| """ | ||
| return [ | ||
| PrivilegeData.from_database_record(record) | ||
| for record in ProviderRecordUtility.get_records_of_type(self.provider_records, ProviderRecordType.PRIVILEGE) | ||
| if (filter_condition is None or filter_condition(PrivilegeData.from_database_record(record))) | ||
| ] | ||
|
|
||
| def get_license_records( | ||
| self, | ||
| filter_condition: Callable[[LicenseData], bool] | None = None, | ||
| ) -> list[LicenseData]: | ||
| """ | ||
| Get all license records from a list of provider records. | ||
| """ | ||
| return [ | ||
| LicenseData.from_database_record(record) | ||
| for record in ProviderRecordUtility.get_records_of_type(self.provider_records, ProviderRecordType.LICENSE) | ||
| if (filter_condition is None or filter_condition(LicenseData.from_database_record(record))) | ||
| ] |
There was a problem hiding this comment.
Optimize record filtering to avoid duplicate conversions.
The current implementation converts each record to a data class twice when a filter is applied - once for the filter check and once for the final list. This is inefficient, especially with large record sets.
Apply this optimization to both methods:
def get_privilege_records(
self,
filter_condition: Callable[[PrivilegeData], bool] | None = None,
) -> list[PrivilegeData]:
"""
Get all privilege records from a list of provider records.
:param filter_condition: An optional filter to apply to the privilege records
"""
- return [
- PrivilegeData.from_database_record(record)
- for record in ProviderRecordUtility.get_records_of_type(self.provider_records, ProviderRecordType.PRIVILEGE)
- if (filter_condition is None or filter_condition(PrivilegeData.from_database_record(record)))
- ]
+ privilege_data_list = [
+ PrivilegeData.from_database_record(record)
+ for record in ProviderRecordUtility.get_records_of_type(self.provider_records, ProviderRecordType.PRIVILEGE)
+ ]
+ if filter_condition:
+ return [privilege for privilege in privilege_data_list if filter_condition(privilege)]
+ return privilege_data_list
def get_license_records(
self,
filter_condition: Callable[[LicenseData], bool] | None = None,
) -> list[LicenseData]:
"""
Get all license records from a list of provider records.
"""
- return [
- LicenseData.from_database_record(record)
- for record in ProviderRecordUtility.get_records_of_type(self.provider_records, ProviderRecordType.LICENSE)
- if (filter_condition is None or filter_condition(LicenseData.from_database_record(record)))
- ]
+ license_data_list = [
+ LicenseData.from_database_record(record)
+ for record in ProviderRecordUtility.get_records_of_type(self.provider_records, ProviderRecordType.LICENSE)
+ ]
+ if filter_condition:
+ return [license for license in license_data_list if filter_condition(license)]
+ return license_data_list📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def get_privilege_records( | |
| self, | |
| filter_condition: Callable[[PrivilegeData], bool] | None = None, | |
| ) -> list[PrivilegeData]: | |
| """ | |
| Get all privilege records from a list of provider records. | |
| :param filter_condition: An optional filter to apply to the privilege records | |
| """ | |
| return [ | |
| PrivilegeData.from_database_record(record) | |
| for record in ProviderRecordUtility.get_records_of_type(self.provider_records, ProviderRecordType.PRIVILEGE) | |
| if (filter_condition is None or filter_condition(PrivilegeData.from_database_record(record))) | |
| ] | |
| def get_license_records( | |
| self, | |
| filter_condition: Callable[[LicenseData], bool] | None = None, | |
| ) -> list[LicenseData]: | |
| """ | |
| Get all license records from a list of provider records. | |
| """ | |
| return [ | |
| LicenseData.from_database_record(record) | |
| for record in ProviderRecordUtility.get_records_of_type(self.provider_records, ProviderRecordType.LICENSE) | |
| if (filter_condition is None or filter_condition(LicenseData.from_database_record(record))) | |
| ] | |
| def get_privilege_records( | |
| self, | |
| filter_condition: Callable[[PrivilegeData], bool] | None = None, | |
| ) -> list[PrivilegeData]: | |
| """ | |
| Get all privilege records from a list of provider records. | |
| :param filter_condition: An optional filter to apply to the privilege records | |
| """ | |
| privilege_data_list = [ | |
| PrivilegeData.from_database_record(record) | |
| for record in ProviderRecordUtility.get_records_of_type( | |
| self.provider_records, | |
| ProviderRecordType.PRIVILEGE | |
| ) | |
| ] | |
| if filter_condition: | |
| return [priv for priv in privilege_data_list if filter_condition(priv)] | |
| return privilege_data_list | |
| def get_license_records( | |
| self, | |
| filter_condition: Callable[[LicenseData], bool] | None = None, | |
| ) -> list[LicenseData]: | |
| """ | |
| Get all license records from a list of provider records. | |
| """ | |
| license_data_list = [ | |
| LicenseData.from_database_record(record) | |
| for record in ProviderRecordUtility.get_records_of_type( | |
| self.provider_records, | |
| ProviderRecordType.LICENSE | |
| ) | |
| ] | |
| if filter_condition: | |
| return [lic for lic in license_data_list if filter_condition(lic)] | |
| return license_data_list |
🤖 Prompt for AI Agents
In
backend/compact-connect/lambdas/python/common/cc_common/data_model/provider_record_util.py
around lines 258 to 283, both get_privilege_records and get_license_records
convert each record twice when a filter_condition is provided, causing
inefficiency. To fix this, refactor each method to convert the record once,
store the converted object in a variable, then apply the filter_condition on
that variable before including it in the result list. This avoids duplicate
conversions and improves performance.
Comment on lines
+1357
to
+1490
| provider_user_records = self.get_provider_user_records( | ||
| compact=compact, | ||
| provider_id=provider_id, | ||
| consistent_read=True, | ||
| ) | ||
|
|
||
| # Get adverse action records for this privilege | ||
| adverse_action_records = provider_user_records.get_adverse_action_records_for_privilege( | ||
| privilege_jurisdiction=jurisdiction, | ||
| privilege_license_type_abbreviation=license_type_abbreviation, | ||
| ) | ||
|
|
||
| # Find the specific adverse action record to lift | ||
| target_adverse_action: AdverseActionData | None = None | ||
| for adverse_action in adverse_action_records: | ||
| if str(adverse_action.adverseActionId) == adverse_action_id: | ||
| target_adverse_action = adverse_action | ||
| break | ||
|
|
||
| if target_adverse_action is None: | ||
| raise CCNotFoundException('Encumbrance record not found') | ||
|
|
||
| # Check if the adverse action has already been lifted | ||
| if target_adverse_action.effectiveLiftDate is not None: | ||
| raise CCInvalidRequestException('Encumbrance has already been lifted') | ||
|
|
||
| # Check if this is the last remaining unlifted adverse action | ||
| unlifted_adverse_actions = [ | ||
| aa | ||
| for aa in adverse_action_records | ||
| if aa.effectiveLiftDate is None and str(aa.adverseActionId) != adverse_action_id | ||
| ] | ||
|
|
||
| # Get the privilege record | ||
| privilege_records = provider_user_records.get_privilege_records( | ||
| filter_condition=lambda p: ( | ||
| p.jurisdiction == jurisdiction | ||
| and p.licenseType | ||
| == LicenseUtility.get_license_type_by_abbreviation(compact, license_type_abbreviation).name | ||
| ) | ||
| ) | ||
|
|
||
| if not privilege_records: | ||
| message = 'Privilege record not found for adverse action record.' | ||
| logger.error(message) | ||
| raise CCInternalException(message) | ||
|
|
||
| privilege_data = privilege_records[0] | ||
|
|
||
| # Build transaction items | ||
| transact_items = [] | ||
|
|
||
| # Always update the adverse action record with lift information | ||
| transact_items.append( | ||
| self._generate_adverse_action_lift_update_item( | ||
| target_adverse_action=target_adverse_action, | ||
| effective_lift_date=effective_lift_date, | ||
| lifting_user=lifting_user, | ||
| ) | ||
| ) | ||
|
|
||
| # If this was the last unlifted adverse action, update privilege status and create update record | ||
| if not unlifted_adverse_actions: | ||
| # Update privilege record to unencumbered status | ||
| privilege_update_item = self._generate_set_privilege_encumbered_status_item( | ||
| privilege_data=privilege_data, | ||
| privilege_encumbered_status=PrivilegeEncumberedStatusEnum.UNENCUMBERED, | ||
| ) | ||
| transact_items.append(privilege_update_item) | ||
|
|
||
| # Create privilege update record | ||
| privilege_update_record = PrivilegeUpdateData.create_new( | ||
| { | ||
| 'type': 'privilegeUpdate', | ||
| 'updateType': UpdateCategory.LIFTING_ENCUMBRANCE, | ||
| 'providerId': provider_id, | ||
| 'compact': compact, | ||
| 'jurisdiction': jurisdiction, | ||
| 'licenseType': privilege_data.licenseType, | ||
| 'previous': privilege_data.to_dict(), | ||
| 'updatedValues': { | ||
| 'encumberedStatus': PrivilegeEncumberedStatusEnum.UNENCUMBERED, | ||
| }, | ||
| } | ||
| ).serialize_to_database_record() | ||
|
|
||
| transact_items.append(self._generate_put_transaction_item(privilege_update_record)) | ||
|
|
||
| # Check if provider should be set to unencumbered | ||
| provider_status_items = self._generate_provider_encumbered_status_transaction_items_if_no_encumbrances( | ||
| provider_user_records=provider_user_records, | ||
| lifted_record=privilege_data, | ||
| ) | ||
| transact_items.extend(provider_status_items) | ||
|
|
||
| # Execute the transaction | ||
| self.config.dynamodb_client.transact_write_items(TransactItems=transact_items) | ||
|
|
||
| logger.info('Successfully lifted privilege encumbrance') | ||
|
|
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add error handling for license type lookup.
The method is well-implemented but needs additional error handling:
- Line 1429:
LicenseUtility.get_license_type_by_abbreviationcould raise an exception if the abbreviation is invalid - Lines 1433-1437: Consider if
CCInternalExceptionis the right exception type - this could be a data consistency issue
Add error handling for the license type lookup:
# Get the privilege record
+ try:
+ license_type_name = LicenseUtility.get_license_type_by_abbreviation(
+ compact, license_type_abbreviation
+ ).name
+ except (KeyError, ValueError) as e:
+ logger.error('Invalid license type abbreviation', error=str(e))
+ raise CCInvalidRequestException(f'Invalid license type abbreviation: {license_type_abbreviation}') from e
+
privilege_records = provider_user_records.get_privilege_records(
filter_condition=lambda p: (
p.jurisdiction == jurisdiction
- and p.licenseType
- == LicenseUtility.get_license_type_by_abbreviation(compact, license_type_abbreviation).name
+ and p.licenseType == license_type_name
)
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def lift_privilege_encumbrance( | |
| self, | |
| compact: str, | |
| provider_id: str, | |
| jurisdiction: str, | |
| license_type_abbreviation: str, | |
| adverse_action_id: str, | |
| effective_lift_date: date, | |
| lifting_user: str, | |
| ) -> None: | |
| """ | |
| Lift an encumbrance from a privilege record by updating the adverse action record | |
| and potentially updating the privilege record's encumbered status. | |
| :param str compact: The compact name | |
| :param str provider_id: The provider ID | |
| :param str jurisdiction: The jurisdiction | |
| :param str license_type_abbreviation: The license type abbreviation | |
| :param str adverse_action_id: The adverse action ID to lift | |
| :param date effective_lift_date: The effective date when the encumbrance is lifted | |
| :param str lifting_user: The cognito sub of the user lifting the encumbrance | |
| :raises CCNotFoundException: If the adverse action record is not found | |
| :raises CCInvalidRequestException: If the encumbrance has already been lifted | |
| """ | |
| with logger.append_context_keys( | |
| compact=compact, | |
| provider_id=provider_id, | |
| jurisdiction=jurisdiction, | |
| license_type_abbreviation=license_type_abbreviation, | |
| adverse_action_id=adverse_action_id, | |
| ): | |
| logger.info('Lifting privilege encumbrance') | |
| # Get all provider records | |
| provider_user_records = self.get_provider_user_records( | |
| compact=compact, | |
| provider_id=provider_id, | |
| consistent_read=True, | |
| ) | |
| # Get adverse action records for this privilege | |
| adverse_action_records = provider_user_records.get_adverse_action_records_for_privilege( | |
| privilege_jurisdiction=jurisdiction, | |
| privilege_license_type_abbreviation=license_type_abbreviation, | |
| ) | |
| # Find the specific adverse action record to lift | |
| target_adverse_action: AdverseActionData | None = None | |
| for adverse_action in adverse_action_records: | |
| if str(adverse_action.adverseActionId) == adverse_action_id: | |
| target_adverse_action = adverse_action | |
| break | |
| if target_adverse_action is None: | |
| raise CCNotFoundException('Encumbrance record not found') | |
| # Check if the adverse action has already been lifted | |
| if target_adverse_action.effectiveLiftDate is not None: | |
| raise CCInvalidRequestException('Encumbrance has already been lifted') | |
| # Check if this is the last remaining unlifted adverse action | |
| unlifted_adverse_actions = [ | |
| aa | |
| for aa in adverse_action_records | |
| if aa.effectiveLiftDate is None and str(aa.adverseActionId) != adverse_action_id | |
| ] | |
| # Get the privilege record | |
| privilege_records = provider_user_records.get_privilege_records( | |
| filter_condition=lambda p: ( | |
| p.jurisdiction == jurisdiction | |
| and p.licenseType | |
| == LicenseUtility.get_license_type_by_abbreviation(compact, license_type_abbreviation).name | |
| ) | |
| ) | |
| if not privilege_records: | |
| message = 'Privilege record not found for adverse action record.' | |
| logger.error(message) | |
| raise CCInternalException(message) | |
| privilege_data = privilege_records[0] | |
| # Build transaction items | |
| transact_items = [] | |
| # Always update the adverse action record with lift information | |
| transact_items.append( | |
| self._generate_adverse_action_lift_update_item( | |
| target_adverse_action=target_adverse_action, | |
| effective_lift_date=effective_lift_date, | |
| lifting_user=lifting_user, | |
| ) | |
| ) | |
| # If this was the last unlifted adverse action, update privilege status and create update record | |
| if not unlifted_adverse_actions: | |
| # Update privilege record to unencumbered status | |
| privilege_update_item = self._generate_set_privilege_encumbered_status_item( | |
| privilege_data=privilege_data, | |
| privilege_encumbered_status=PrivilegeEncumberedStatusEnum.UNENCUMBERED, | |
| ) | |
| transact_items.append(privilege_update_item) | |
| # Create privilege update record | |
| privilege_update_record = PrivilegeUpdateData.create_new( | |
| { | |
| 'type': 'privilegeUpdate', | |
| 'updateType': UpdateCategory.LIFTING_ENCUMBRANCE, | |
| 'providerId': provider_id, | |
| 'compact': compact, | |
| 'jurisdiction': jurisdiction, | |
| 'licenseType': privilege_data.licenseType, | |
| 'previous': privilege_data.to_dict(), | |
| 'updatedValues': { | |
| 'encumberedStatus': PrivilegeEncumberedStatusEnum.UNENCUMBERED, | |
| }, | |
| } | |
| ).serialize_to_database_record() | |
| transact_items.append(self._generate_put_transaction_item(privilege_update_record)) | |
| # Check if provider should be set to unencumbered | |
| provider_status_items = self._generate_provider_encumbered_status_transaction_items_if_no_encumbrances( | |
| provider_user_records=provider_user_records, | |
| lifted_record=privilege_data, | |
| ) | |
| transact_items.extend(provider_status_items) | |
| # Execute the transaction | |
| self.config.dynamodb_client.transact_write_items(TransactItems=transact_items) | |
| logger.info('Successfully lifted privilege encumbrance') | |
| # Get the privilege record | |
| try: | |
| license_type_name = LicenseUtility.get_license_type_by_abbreviation( | |
| compact, license_type_abbreviation | |
| ).name | |
| except (KeyError, ValueError) as e: | |
| logger.error('Invalid license type abbreviation', error=str(e)) | |
| raise CCInvalidRequestException( | |
| f'Invalid license type abbreviation: {license_type_abbreviation}' | |
| ) from e | |
| privilege_records = provider_user_records.get_privilege_records( | |
| filter_condition=lambda p: ( | |
| p.jurisdiction == jurisdiction | |
| and p.licenseType == license_type_name | |
| ) | |
| ) |
🤖 Prompt for AI Agents
In
backend/compact-connect/lambdas/python/common/cc_common/data_model/data_client.py
around lines 1429 to 1437, add error handling for the call to
LicenseUtility.get_license_type_by_abbreviation to catch exceptions if the
license type abbreviation is invalid. Wrap this call in a try-except block and
raise a more specific exception or handle the error gracefully. Also, review the
use of CCInternalException when no privilege records are found and consider
raising a different exception that better reflects a data consistency or
validation issue.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
State admins need the ability to lift an encumbrance from a license or privilege when appropriate. This adds new endpoints for setting an effective lift date on license and privilege encumbrances.
Closes #678
Summary by CodeRabbit
New Features
Bug Fixes
Documentation
Tests
Refactor