### Install osdu-client using pip Source: https://github.com/micmurawski/osdu-client/blob/main/README.md This command installs the osdu-client library from PyPI. It's the standard way to add the library to your Python project. ```bash pip install osdu-client ``` -------------------------------- ### Initialize OSDU Service Clients using OSDUAPI Factory Source: https://context7.com/micmurawski/osdu-client/llms.txt Shows how to create service-specific clients for the OSDU platform using the OSDUAPI.client() factory method. It covers creating clients for storage and search services, disabling validation, and selecting specific API versions. The example also demonstrates listing available services. ```python from osdu_client import OSDUAPI from osdu_client.auth import AuthBackendInterface class AuthBackend(AuthBackendInterface): authorization_header = {"Authorization": "Bearer token123"} default_data_partition_id = "opendes" base_url = "https://osdu.example.com" def get_sd_connection_params(self, log_level: int = None) -> dict: return {} auth_backend = AuthBackend() # Create storage service client storage_client = OSDUAPI.client('storage', auth_backend=auth_backend) # Create search service client with validation disabled search_client = OSDUAPI.client('search', auth_backend=auth_backend, validation=False) # Create versioned service client (e.g., RAFS v2) rafs_client = OSDUAPI.client('rafs', auth_backend=auth_backend, version='v2') # List all available services OSDUAPI.print_available_services() # Output: # | Name | Versions | # =================================== # | storage | latest | # | search | latest | # | rafs | v1, v2 | ``` -------------------------------- ### Create and Update OSDU Records using Python Source: https://context7.com/micmurawski/osdu-client/llms.txt Provides examples for creating new OSDU records and updating existing ones, including metadata like ACL and legal tags. It also demonstrates patching records to modify metadata and handles potential OSDUAPIError exceptions. ```python storage_client = OSDUAPI.client('storage', auth_backend=auth_backend) # Create or update a record new_record = { "id": "opendes:work-product-component:seismic-123", "kind": "opendes:wks:work-product-component--SeismicTraceData:1.0.0", "acl": { "viewers": ["data.default.viewers@opendes.example.com"], "owners": ["data.default.owners@opendes.example.com"] }, "legal": { "legaltags": ["opendes-public-usa-dataset"], "otherRelevantDataCountries": ["US"] }, "data": { "Name": "Seismic Survey Q1 2024", "Description": "3D seismic data acquisition", "SurveyArea": "Gulf of Mexico", "AcquisitionDate": "2024-03-15" } } try: result = storage_client.update_records() print(f"Record created: {result['recordIds']}") # Patch update multiple records (metadata only) patch_result = storage_client.patch_records( query={"ids": ["opendes:wellbore:123456"]}, ops=[ { "op": "replace", "path": "/legal/legaltags", "value": ["opendes-public-usa-dataset", "opendes-confidential"] } ] ) print(f"Records updated: {patch_result['recordCount']}") except OSDUAPIError as e: print(f"Failed to update record: {e.message}") ``` -------------------------------- ### File Service: Download, Manage Metadata, and Delete Files with OSDU Client Source: https://context7.com/micmurawski/osdu-client/llms.txt This Python snippet demonstrates how to use the OSDU File Service client to retrieve file metadata, obtain signed download URLs, download files using requests, delete file metadata, and get file location information. It includes error handling for OSDUAPIError. ```python file_client = OSDUAPI.client('file', auth_backend=auth_backend) try: # Get file metadata metadata = file_client.get_files_metadata( id="opendes:dataset--File.Generic:12345" ) print(f"File name: {metadata['data']['DatasetProperties']['FileSourceInfo']['Name']}") # Get download URL with custom expiration download_url_response = file_client.gets_url_to_download_file( id="opendes:dataset--File.Generic:12345", expiry_time="24H" # Valid for 24 hours ) signed_download_url = download_url_response['signedUrl'] print(f"Download URL: {signed_download_url}") # Download file using signed URL import requests response = requests.get(signed_download_url) with open("downloaded_seismic_data.segy", "wb") as f: f.write(response.content) print("File downloaded successfully") # Delete file and metadata delete_result = file_client.delete_files_metadata( id="opendes:dataset--File.Generic:12345" ) print(f"File and metadata deleted") # Get file location information location_info = file_client.get_file_location( file_id="opendes:dataset--File.Generic:12345" ) print(f"Location: {location_info['Location']}") print(f"Driver: {location_info['Driver']}") except OSDUAPIError as e: print(f"File download/management failed: {e.message}") ``` -------------------------------- ### Manage Group Memberships with Entitlements Service Client (Python) Source: https://context7.com/micmurawski/osdu-client/llms.txt This Python snippet demonstrates how to use the OSDU API client for the Entitlements service to manage group memberships. It covers operations like getting a member's groups, counting members in a group, removing a member, listing partition groups, and updating group properties. It requires an authenticated `OSDUAPI` client instance and handles potential `OSDUAPIError` exceptions. ```python entitlements_client = OSDUAPI.client('entitlements', auth_backend=auth_backend) try: # Get all groups for a specific member member_groups = entitlements_client.get_members_groups( member_email="user@example.com", type="DATA", # DATA, SERVICE, or USER groups role_required="true" ) print(f"Member belongs to {len(member_groups['groups'])} groups") for group in member_groups['groups']: print(f"Group: {group['email']}, Role: {group['role']}") # Get member count for a group member_count = entitlements_client.get_count_group_members( group_email="seismic.viewers@opendes.example.com", role="MEMBER" ) print(f"Group has {member_count['memberCount']} members") # Remove member from specific group remove_result = entitlements_client.delete_member_from_group( group_email="seismic.viewers@opendes.example.com", member_email="user@example.com" ) print("Member removed from group") # List all groups in partition all_partition_groups = entitlements_client.list_partition_groups( type="DATA", limit=100, cursor=None ) print(f"Partition has {len(all_partition_groups['groups'])} groups") # Update group properties update_result = entitlements_client.update_groups( group_email="seismic.viewers@opendes.example.com", update_group_request=[ { "op": "replace", "path": "/description", "value": "Updated description for seismic viewers group" } ] ) except OSDUAPIError as e: print(f"Member operation failed: {e.message}") ``` -------------------------------- ### Search Service - Query Records (Python) Source: https://context7.com/micmurawski/osdu-client/llms.txt Demonstrates how to search for records using full-text queries and cursor-based pagination. It requires an initialized OSDUAPI client for the 'search' service and handles potential OSDUAPIErrors. ```python from osdu_client import OSDUAPI search_client = OSDUAPI.client('search', auth_backend=auth_backend) try: # Full-text search with filters search_results = search_client.query( kind={"*": ["opendes:wks:*:*"]}, query="Gulf of Mexico AND seismic", limit=100, offset=0, returned_fields=["data.Name", "data.AcquisitionDate"], sort={ "field": ["data.AcquisitionDate"], "order": ["DESC"] }, track_total_count=True ) print(f"Total results: {search_results['totalCount']}") for result in search_results['results']: print(f"ID: {result['id']}, Name: {result['data']['Name']}") # Cursor-based pagination for large result sets cursor_results = search_client.query_with_cursor( kind={"*": ["opendes:wks:master-data--Well:*"]}, query="status:ACTIVE", limit=1000, cursor=None # First page ) print(f"Retrieved {len(cursor_results['results'])} results") next_cursor = cursor_results.get('cursor') # Fetch next page if next_cursor: next_page = search_client.query_with_cursor( kind={"*": ["opendes:wks:master-data--Well:*"]}, query="status:ACTIVE", limit=1000, cursor=next_cursor ) except OSDUAPIError as e: print(f"Search failed: {e.message}") ``` -------------------------------- ### Implement Custom Authentication Backend for OSDU API Source: https://context7.com/micmurawski/osdu-client/llms.txt Demonstrates how to implement a custom authentication backend by inheriting from AuthBackendInterface. This involves providing authorization headers, default data partition ID, base URL, and optional SD connection parameters for OSDU API access. ```python from osdu_client.auth import AuthBackendInterface class CustomAuthBackend(AuthBackendInterface): def __init__(self, access_token: str, base_url: str, partition_id: str): self._access_token = access_token self._base_url = base_url self._partition_id = partition_id @property def authorization_header(self) -> dict: return {"Authorization": f"Bearer {self._access_token}"} @property def default_data_partition_id(self) -> str: return self._partition_id @property def base_url(self) -> str: return self._base_url def get_sd_connection_params(self, log_level: int = None) -> dict: return {} # Initialize authentication backend auth = CustomAuthBackend( access_token="eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...", base_url="https://osdu.example.com", partition_id="opendes" ) ``` -------------------------------- ### Upload File and Create Metadata (Python) Source: https://context7.com/micmurawski/osdu-client/llms.txt Details the three-step process for uploading a file using the OSDU file service: obtaining a signed URL, uploading the file content, and then creating a metadata record for the uploaded file. Requires the OSDUAPI client, requests library, and authentication backend. ```python from osdu_client import OSDUAPI import requests file_client = OSDUAPI.client('file', auth_backend=auth_backend) try: # Step 1: Get signed upload URL upload_url_response = file_client.get_files_upload_url() signed_url = upload_url_response['signedUrl'] file_source = upload_url_response['FileSource'] print(f"Upload URL: {signed_url}") print(f"File source: {file_source}") # Step 2: Upload file to signed URL with open("/path/to/seismic_data.segy", "rb") as file: upload_response = requests.put(signed_url, data=file) upload_response.raise_for_status() print("File uploaded successfully") # Step 3: Create file metadata record metadata_record = file_client.create_files_metadata( kind="opendes:wks:dataset--File.Generic:1.0.0", acl={ "viewers": ["data.default.viewers@opendes.example.com"], "owners": ["data.default.owners@opendes.example.com"] }, legal={ "legaltags": ["opendes-public-usa-seismic"], "otherRelevantDataCountries": ["US"] }, data={ "Name": "Seismic Survey Data Q1 2024", "DatasetProperties": { "FileSourceInfo": { "FileSource": file_source, "Name": "seismic_data.segy" } } } ) print(f"File metadata created: {metadata_record['id']}") except OSDUAPIError as e: print(f"File operation failed: {e.message}") ``` -------------------------------- ### Schema Management with OSDU Python Client Source: https://context7.com/micmurawski/osdu-client/llms.txt This snippet illustrates how to manage data schemas using the OSDU Python client. It covers creating a new schema with specified identity, status, scope, and properties, and then retrieving an existing schema by its ID. Schemas follow a versioning model and define data structure and validation rules. Requires an initialized `OSDUAPI` client with an authentication backend. ```python from osdu_client import OSDUAPI schema_client = OSDUAPI.client('schema', auth_backend=auth_backend) try: # Create a new schema new_schema = schema_client.create_schema( schema_info={ "schemaIdentity": { "authority": "opendes", "source": "wks", "entityType": "CustomWellData", "schemaVersionMajor": 1, "schemaVersionMinor": 0, "schemaVersionPatch": 0 }, "status": "PUBLISHED", "scope": "INTERNAL" }, schema={ "type": "object", "properties": { "WellName": {"type": "string"}, "Operator": {"type": "string"}, "SpudDate": {"type": "string", "format": "date"}, "TotalDepth": {"type": "number"}, "WellStatus": { "type": "string", "enum": ["DRILLING", "COMPLETED", "ABANDONED"] } }, "required": ["WellName", "Operator"] } ) print(f"Schema created: {new_schema['schemaIdentity']['id']}") # Get schema by ID schema = schema_client.get_schema( id="opendes:wks:CustomWellData:1.0.0" ) print(f"Schema: {schema['schemaIdentity']['entityType']}") print(f"Properties: {list(schema['schema']['properties'].keys())}") except OSDUAPIError as e: print(f"Schema operation failed: {e.message}") ``` -------------------------------- ### Print Available OSDU Services Source: https://github.com/micmurawski/osdu-client/blob/main/README.md This Python code snippet shows how to print a list of all available services provided by the OSDU API. It utilizes the OSDUAPI client to access this information. ```python from osdu_client.client import OSDUAPI OSDUAPI.print_available_services() ``` -------------------------------- ### Initialize OSDU API Client with Custom Auth Source: https://github.com/micmurawski/osdu-client/blob/main/README.md This Python code demonstrates how to initialize the OSDU API client with a custom authentication backend. It defines a session class inheriting from AuthBackendInterface, sets base URL and authorization headers, and then creates a storage client instance to interact with the OSDU storage service. ```python from osdu_client import OSDUAPI from osdu_client.auth import AuthBackendInterface class AuthSession(AuthBackendInterface): base_url = "https://base.url" default_data_partition_id = "osdu" authorization_header = {"Authorization": "Bearer access_token"} def get_sd_connection_params(self): return {} auth_backend = AuthSession( headers={"Authorization": "Bearer XYZ"}, base_url="https://exmaple.com" ) storage_client = OSDUAPI.client('storage', auth_backend=auth_backend) response = storage_client.get_record_versions(id="123") ``` -------------------------------- ### Search Service - Advanced Queries (Python) Source: https://context7.com/micmurawski/osdu-client/llms.txt Illustrates advanced search capabilities including spatial filtering (bounding box), aggregations for faceted results, and range queries on dates. It utilizes the 'search' client and expects an authenticated session. ```python search_client = OSDUAPI.client('search', auth_backend=auth_backend) # Spatial search with bounding box spatial_results = search_client.query( kind={"*": ["opendes:wks:master-data--Wellbore:*"]}, spatial_filter={ "field": "data.SpatialLocation", "byBoundingBox": { "topLeft": { "latitude": 29.5, "longitude": -95.5 }, "bottomRight": { "latitude": 28.5, "longitude": -94.5 } } }, limit=50 ) # Aggregation query for faceted results agg_results = search_client.query( kind={"*": ["opendes:wks:work-product-component--*:*"]}, query="*", aggregate_by="data.ComponentType", limit=0 # Only aggregations, no records ) print("Component type distribution:") for bucket in agg_results['aggregations']: print(f"{bucket['key']}: {bucket['count']} records") # Range query with date filtering date_range_results = search_client.query( kind={"*": ["opendes:wks:*:*"]}, query="data.AcquisitionDate:[2024-01-01 TO 2024-12-31]", returned_fields=["data.Name", "data.AcquisitionDate"], sort={"field": ["data.AcquisitionDate"], "order": ["ASC"]}, limit=100 ) ``` -------------------------------- ### Legal Service - Legal Tag Management (Python) Source: https://context7.com/micmurawski/osdu-client/llms.txt Shows how to create, list, retrieve, and update legal tags using the 'legal' client. This is essential for data governance and compliance within OSDU, and includes error handling for OSDUAPIError. ```python from osdu_client import OSDUAPI legal_client = OSDUAPI.client('legal', auth_backend=auth_backend) try: # Create a new legal tag new_tag = legal_client.create_legaltag( name="opendes-public-usa-seismic", description="Public seismic data acquired in USA waters", properties={ "contractId": "CONTRACT-2024-001", "countryOfOrigin": ["US"], "dataType": "Public Domain Data", "expirationDate": "2034-12-31", "exportClassification": "EAR99", "originator": "Example Energy Corp", "personalData": "No Personal Data", "securityClassification": "Public" } ) print(f"Legal tag created: {new_tag['name']}") # List all legal tags all_tags = legal_client.list_legaltags(valid=True) print(f"Total legal tags: {len(all_tags['legalTags'])}") # Get specific legal tag tag_details = legal_client.get_legaltag( name="opendes-public-usa-seismic" ) print(f"Tag: {tag_details['name']}, Status: {tag_details['status']}") # Update legal tag properties updated_tag = legal_client.update_legaltag( name="opendes-public-usa-seismic", description="Updated description with extended coverage", expiration_date="2035-12-31" ) except OSDUAPIError as e: print(f"Legal tag operation failed: {e.message}") ``` -------------------------------- ### Search and Update OSDU Schemas (Python) Source: https://context7.com/micmurawski/osdu-client/llms.txt Demonstrates searching for OSDU schemas using filters and updating schemas that are in the DEVELOPMENT status. Published schemas are immutable. Requires the OSDUAPI client and authentication backend. ```python schema_client = OSDUAPI.client('schema', auth_backend=auth_backend) try: # Search for schemas with filters search_results = schema_client.search_schemas( authority="opendes", source="wks", entity_type="CustomWellData", status="PUBLISHED", latest_version="true", limit="100" ) print(f"Found {len(search_results['schemaInfos'])} schemas") for schema_info in search_results['schemaInfos']: identity = schema_info['schemaIdentity'] version = f"{identity['schemaVersionMajor']}.{identity['schemaVersionMinor']}.{identity['schemaVersionPatch']}" print(f"{identity['entityType']} v{version} - {schema_info['status']}") # Update schema in DEVELOPMENT status updated_schema = schema_client.update_schema( schema_info={ "schemaIdentity": { "authority": "opendes", "source": "wks", "entityType": "CustomWellData", "schemaVersionMajor": 1, "schemaVersionMinor": 1, "schemaVersionPatch": 0 }, "status": "DEVELOPMENT" }, schema={ "type": "object", "properties": { "WellName": {"type": "string"}, "Operator": {"type": "string"}, "Location": { "type": "object", "properties": { "Latitude": {"type": "number"}, "Longitude": {"type": "number"} } } } } ) except OSDUAPIError as e: print(f"Schema search/update failed: {e.message}") ``` -------------------------------- ### Entitlements Service: Manage Groups and Members with OSDU Client Source: https://context7.com/micmurawski/osdu-client/llms.txt This Python snippet illustrates how to interact with the OSDU Entitlements Service to manage user groups and permissions. It covers listing user groups, creating new data groups, adding members to groups, and retrieving group members. Error handling for OSDUAPIError is included. ```python from osdu_client import OSDUAPI entitlements_client = OSDUAPI.client('entitlements', auth_backend=auth_backend) try: # List all groups user belongs to my_groups = entitlements_client.get_groups() print(f"User is member of {len(my_groups['groups'])} groups") for group in my_groups['groups']: print(f"Group: {group['email']}, Role: {group['role']}") # Create a new data group new_group = entitlements_client.create_group( group_info_dto={ "name": "seismic.viewers", "description": "Read access to seismic data", "email": "seismic.viewers@opendes.example.com" } ) print(f"Group created: {new_group['email']}") # Add member to group add_result = entitlements_client.add_member( group_email="seismic.viewers@opendes.example.com", add_member_dto={ "email": "user@example.com", "role": "MEMBER" # or "OWNER" } ) print(f"Member added to group") # Get group members members = entitlements_client.get_groups_members( group_email="seismic.viewers@opendes.example.com", role="MEMBER" ) print(f"Group has {len(members['members'])} members") except OSDUAPIError as e: print(f"Entitlements operation failed: {e.message}") ``` -------------------------------- ### Retrieve OSDU Records using Python Source: https://context7.com/micmurawski/osdu-client/llms.txt Demonstrates how to retrieve single records, records with filtered attributes, multiple records, and all versions of a record using the osdu-client library. Handles potential OSDUAPIError exceptions during retrieval. ```python from osdu_client import OSDUAPI from osdu_client.exceptions import OSDUAPIError storage_client = OSDUAPI.client('storage', auth_backend=auth_backend) try: # Get a single record by ID record = storage_client.get_record( id="opendes:wellbore:123456" ) print(f"Record ID: {record['id']}") print(f"Kind: {record['kind']}") print(f"Data: {record['data']}") # Get record with filtered attributes filtered_record = storage_client.get_record( id="opendes:wellbore:123456", attribute=["data.WellboreName", "data.WellID"] ) # Get multiple records at once batch_records = storage_client.query_records( records=[ "opendes:wellbore:123456", "opendes:wellbore:789012" ] ) print(f"Retrieved {len(batch_records['records'])} records") # Get all record versions versions = storage_client.get_record_versions( id="opendes:wellbore:123456" ) print(f"Available versions: {versions['versions']}") except OSDUAPIError as e: print(f"API Error: {e.message}, Status: {e.status_code}") ``` -------------------------------- ### Batch Legal Tag Operations with OSDU Python Client Source: https://context7.com/micmurawski/osdu-client/llms.txt This snippet demonstrates how to use the OSDU Python client to perform batch operations on legal tags. It covers retrieving multiple tags at once, validating their compliance, and querying tags with filters. The maximum number of tags per request is 25. Requires an initialized `OSDUAPI` client with an authentication backend. ```python legal_client = OSDUAPI.client('legal', auth_backend=auth_backend) try: # Retrieve multiple legal tags at once tag_names = [ "opendes-public-usa-seismic", "opendes-confidential-exploration", "opendes-restricted-production" ] batch_tags = legal_client.get_batch_legaltags(names=tag_names) print(f"Retrieved {len(batch_tags['legalTags'])} tags") for tag in batch_tags['legalTags']: print(f"Tag: {tag['name']}, Valid: {tag['status'] == 'compliant'}") # Validate legal tags validation_result = legal_client.validate_legaltags(names=tag_names) for tag in validation_result['legalTags']: if tag['status'] != 'compliant': print(f"Invalid tag {tag['name']}: {tag['invalidReason']}") # Query legal tags with filters query_results = legal_client.query_legaltags( valid=True, query_list=["properties.countryOfOrigin:US"], sort_by="name", sort_order="ASC", limit=50 ) print(f"Found {len(query_results['legalTags'])} USA legal tags") except OSDUAPIError as e: print(f"Batch operation failed: {e.message}") ``` -------------------------------- ### Delete OSDU Records using Python Source: https://context7.com/micmurawski/osdu-client/llms.txt Illustrates how to perform soft and hard deletions of OSDU records, as well as purging specific record versions. It covers single and bulk deletion operations and includes error handling for OSDUAPIError. ```python storage_client = OSDUAPI.client('storage', auth_backend=auth_backend) try: # Soft delete (logical deletion, can be reverted) delete_result = storage_client.delete_record( id="opendes:wellbore:123456" ) print(f"Record soft deleted: {delete_result}") # Soft delete multiple records bulk_delete = storage_client.create_records_delete() # Purge record permanently (cannot be undone) purge_result = storage_client.purge_record( id="opendes:wellbore:123456" ) print(f"Record permanently deleted") # Delete old record versions version_purge = storage_client.purge_record_versions( id="opendes:wellbore:123456", limit=10 # Delete oldest 10 versions ) print(f"Deleted {version_purge['deletedVersions']} versions") except OSDUAPIError as e: print(f"Deletion failed: {e.message}") ``` -------------------------------- ### Handle OSDU API Errors with Specific Exception Types (Python) Source: https://context7.com/micmurawski/osdu-client/llms.txt This Python code demonstrates robust error handling for OSDU API interactions using the `osdu_client` library. It shows how to catch `OSDUAPIError` for API-level issues (like 404, 403, 401, 5XX errors) and `OSDUClientError` for client configuration problems. It also illustrates disabling request validation to allow malformed requests to be handled by the API. ```python from osdu_client import OSDUAPI from osdu_client.exceptions import OSDUAPIError, OSDUClientError try: storage_client = OSDUAPI.client('storage', auth_backend=auth_backend) # Attempt to retrieve non-existent record record = storage_client.get_record(id="invalid:record:id") except OSDUAPIError as e: # API returned 4XX or 5XX error print(f"API Error: {e.message}") print(f"Status Code: {e.status_code}") if e.status_code == 404: print("Record not found") elif e.status_code == 403: print("Access denied - check permissions") elif e.status_code == 401: print("Authentication failed - check token") elif e.status_code >= 500: print("Server error - retry later") except OSDUClientError as e: # Client configuration error (invalid service, version, etc.) print(f"Client Error: {str(e)}") # Disable validation for performance try: storage_client = OSDUAPI.client( 'storage', auth_backend=auth_backend, validation=False # Skip request body validation ) # Malformed request will fail at API level instead of client validation storage_client.update_records() except OSDUAPIError as e: print(f"API rejected invalid request: {e.message}") ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.