yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-07-25 18:30:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

Ingestion: Refactor, docs, airflow example

Browse Source
This commit is contained in:
Suresh Srinivas 2021-08-14 11:43:53 -07:00
parent 8256de30e9
commit 5191cc2dca
3 changed files with 18 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
ingestion/src/metadata/ingestion/sink/metadata_rest_users.py
View File

@ -20,7 +20,7 @@ from metadata.ingestion.api.common import WorkflowContext, Record
from metadata.ingestion.api.sink import Sink, SinkStatus from metadata.ingestion.api.sink import Sink, SinkStatus
from metadata.ingestion.models.user import MetadataTeam, MetadataUser from metadata.ingestion.models.user import MetadataTeam, MetadataUser
from metadata.ingestion.ometa.auth_provider import MetadataServerConfig from metadata.ingestion.ometa.auth_provider import MetadataServerConfig
from metadata.ingestion.ometa.client import REST from metadata.ingestion.ometa.client import REST, APIError
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -62,16 +62,19 @@ class MetadataRestUsersSink(Sink):
team_response = self.rest.get(self.api_team_get) team_response = self.rest.get(self.api_team_get)
for team in team_response['data']: for team in team_response['data']:
self.team_entities[team['name']] = team['id'] self.team_entities[team['displayName']] = team['id']
def _create_team(self, record: MetadataUser) -> None: def _create_team(self, record: MetadataUser) -> None:
team_name = record.team_name team_name = record.team_name
metadata_team = MetadataTeam(team_name, 'Team Name') metadata_team = MetadataTeam(team_name, 'Team Name')
r = self.rest.post(self.api_team_post, try:
data=metadata_team.to_json() r = self.rest.post(self.api_team_post,
) data=metadata_team.to_json()
instance_id = r['id'] )
self.team_entities[team_name] = instance_id instance_id = r['id']
self.team_entities[team_name] = instance_id
except APIError:
pass
def _create_user(self, record: MetadataUser) -> None: def _create_user(self, record: MetadataUser) -> None:
if record.team_name not in self.team_entities: if record.team_name not in self.team_entities:
@ -82,9 +85,12 @@ class MetadataRestUsersSink(Sink):
display_name=record.name, display_name=record.name,
email=record.email, email=record.email,
teams=teams) teams=teams)
self.rest.post(self.api_users, data=metadata_user.to_json()) try:
self.status.records_written(record.github_username) self.rest.post(self.api_users, data=metadata_user.to_json())
logger.info("Sink: {}".format(record.github_username)) self.status.records_written(record.github_username)
logger.info("Sink: {}".format(record.github_username))
except APIError:
pass
def get_status(self) -> SinkStatus: def get_status(self) -> SinkStatus:
return self.status return self.status

2
ingestion/src/metadata/ingestion/source/sample_users.py
View File

@ -53,7 +53,7 @@ class SampleUserMetadataGenerator:
schema['full_name'] = lambda: None schema['full_name'] = lambda: None
schema['github_username'] = lambda: None schema['github_username'] = lambda: None
schema['team_name'] = lambda: random.choice( schema['team_name'] = lambda: random.choice(
['Data_Infra', 'Infra', 'Payments', 'Legal', 'Dev_Platform', 'Trust', 'Marketplace']) ['Data Platform', 'Cloud Infra', 'Payments', 'Legal', 'Customer Support', 'Finance', 'Marketplace'])
schema['employee_type'] = lambda: None schema['employee_type'] = lambda: None
schema['manager_email'] = lambda: fake.email() schema['manager_email'] = lambda: fake.email()
schema['slack_id'] = lambda: None schema['slack_id'] = lambda: None

2
ingestion/src/metadata/ingestion/source/sql_source.py
View File

@ -234,9 +234,9 @@ class SQLSource(Source):
db = DatabaseEntity(id=uuid.uuid4(), db = DatabaseEntity(id=uuid.uuid4(),
name=schema, name=schema,
description=description if description is not None else ' ',
service=EntityReference(id=self.service.id, type=self.config.service_type)) service=EntityReference(id=self.service.id, type=self.config.service_type))
table = TableEntity(name=table, table = TableEntity(name=table,
description=description if description is not None else ' ',
columns=table_columns) columns=table_columns)
table_and_db = OMetaDatabaseAndTable(table=table, database=db) table_and_db = OMetaDatabaseAndTable(table=table, database=db)