def createLineage()

in code/lambda_layer/metadata-services/python/datastore.py [0:0]
42 lines of code
7 McCabe index (conditional complexity)

    def createLineage(self, documentId, callerId, targetBucketName, targetFileName, timestamp, s3Event, sourceBucketName=None, sourceFileName=None, versionId=None):
        ret = None
        
        dynamodb = AwsHelper().getResource("dynamodb")
        table = dynamodb.Table(self._lineageTableName)
        documentSignature = "BUCKET:{}@FILE:{}".format(targetBucketName, targetFileName)
        if versionId:
            documentSignature += "@VERSION:{}".format(versionId)
        item = {
            "documentId": documentId,
            "documentSignature": documentSignature,
            "callerId": callerId,
            "targetBucketName": targetBucketName,
            "targetFileName": targetFileName,
            "timestamp": timestamp,
            "s3Event": s3Event
        }
        if versionId:
            item['versionId'] = versionId
        if sourceFileName:
            item['sourceFileName'] = sourceFileName
        if sourceBucketName:
            item['sourceBucketName'] = sourceBucketName
        try:
            table.put_item(
                Item = item
            )
            ret = {
                'Status': 200
            }
        except ClientError as e:
            print(e)
            ret = {
                'Error': e.response['Error']['Message'],
                'Status': e.response['ResponseMetadata']['HTTPStatusCode']
            }
        except Exception as e:
            print(e)
            ret = {
                'Error': 'Unknown error occurred during updating document',
                'Status': 400
            }
        return ret