https://datahubproject.io logo
Join Slack
Powered by
def add_owners(gms_endpoint, owner_id, owner_type,...
# getting-started
l
def add_owners(gms_endpoint, owner_id, owner_type,entity_urn): need_write = False emitter = DatahubRestEmitter(gms_endpoint) owner_to_add = make_user_urn(owner_id) ownership_type = ownershiptype[owner_type] owner_class_to_add = get_owner_class_to_add(owner_id, owner_type) ownership_to_add = OwnershipClass(owners=[owner_class_to_add]) current_owners = get_current_owners(gms_endpoint=gms_endpoint,entity_urn=entity_urn) if current_owners: if( (owner_to_add, ownership_type) not in [ (x.owner, x.type) for x in current_owners.owners]): # owners exist, but this owner is not present in the current owners current_owners.owners.append(ownership_to_add) need_write = True else: # create a brand new ownership aspect current_owners = ownership_to_add need_write = True if need_write: event: MetadataChangeProposalWrapper = MetadataChangeProposalWrapper( entityUrn=entity_urn, aspect=current_owners, ) emitter.emit_mcp(event) log.info( f"Owner {owner_to_add}, type {ownership_type} added to dataset {entity_urn}" ) else: log.info(f"Owner {owner_to_add} already exists, omitting write")
r
Hi Sheela, can you share more information about what version you're running & what errors you're hitting? That will help folks in the Community help you debug more effectively 🙂
l
I am getting this schema error
File "C:\Users\sheelak\AppData\Local\Programs\Python\Python311\Lib\site-packages\avrogen\avrojson.py", line 131, in to_json_object raise AvroTypeException(writers_schema, data_obj) avro.io.AvroTypeException: The datum OwnershipClass({'owners': [OwnershipClass({'owners': [OwnerClass({'owner': 'urnlicorpuser:asaraswat', 'type': 'TECHNICAL_OWNER', 'typeUrn': None, 'source': None})], 'lastModified': AuditStampClass({'time': 0, 'actor': 'urnlicorpuser:unknown', 'impersonator': None, 'message': None})})], 'lastModified': AuditStampClass({'time': 0, 'actor': 'urnlicorpuser:k8s', 'impersonator': None, 'message': None})}) is not an example of the schema {
if I pass one ownershipclass ,it works. if I pass current_owners with multiple owners , it wont work, throws schema error
Copy code
def add_owners(gms_endpoint, owner_id, owner_type,entity_urn):
    need_write = False
    emitter = DatahubRestEmitter(gms_endpoint)
    owner_to_add = make_user_urn(owner_id)
    ownership_type = ownershiptype[owner_type]
    owner_class_to_add = get_owner_class_to_add(owner_id, owner_type)
    ownership_to_add = OwnershipClass(owners=[owner_class_to_add])
    current_owners = get_current_owners(gms_endpoint=gms_endpoint,entity_urn=entity_urn)
    if current_owners:
        for owner in current_owners.owners:
            if((owner_to_add, ownership_type) not in (owner.owner, owner.type)):
                current_owners.owners.remove(owner)
                current_owners.owners.append(ownership_to_add)
                break
        need_write = True
    else:
        # create a brand new ownership aspect
        current_owners = ownership_to_add
        need_write = True

    if need_write:
        event: MetadataChangeProposalWrapper = MetadataChangeProposalWrapper(
        entityUrn=entity_urn,
        aspect=current_owners,
    )
        emitter.emit_mcp(event)
        <http://log.info|log.info>(
        f"Owner {owner_to_add}, type {ownership_type} added to dataset {entity_urn}"
        )
    else:
        <http://log.info|log.info>(f"Owner {owner_to_add} already exists, omitting write")
this is the code I am using to update the ownersip of one user among the current users
getting schema error as seen above
if I pass ownershipClass with single owner, existing owners are replaced with the passed owner
Hi all can any one check this and suggest me , if there is any workaround for this ?
e
Hello, we are facing same type of error, is there any workaround this??
l
It works
Copy code
def add_owners(gms_endpoint, owner_id, owner_type,entity_urn):
    need_write = False
    emitter = DatahubRestEmitter(gms_endpoint)
    owner_to_add = make_user_urn(owner_id)
    ownership_type = ownershiptype[owner_type]
    owner_class_to_add = get_owner_class_to_add(owner_id, owner_type)
    ownership_to_add = OwnershipClass(owners=[owner_class_to_add])
    current_owners = get_current_owners(gms_endpoint=gms_endpoint,entity_urn=entity_urn)
    if current_owners.owners:
        for owner in current_owners.owners:
            if((owner_to_add, ownership_type) not in (owner.owner, owner.type)):
                current_owners.owners.append(owner_class_to_add)
                break
        need_write = True
    else:
        # create a brand new ownership aspect
        current_owners = ownership_to_add
        need_write = True

    if need_write:
        event: MetadataChangeProposalWrapper = MetadataChangeProposalWrapper(
        entityUrn=entity_urn,
        aspect=current_owners,
    )
        emitter.emit_mcp(event)
        <http://log.info|log.info>(
        f"Owner {owner_to_add}, type {ownership_type} added to entity {entity_urn}"
        )
    else:
        <http://log.info|log.info>(f"Owner {owner_to_add} already exists, omitting write")
The above code worked for me
Open in Slack
PreviousNext
Powered by