Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_full_pii(self):
pii_table = GlueStore.get_pii_table(MockExplorer.get_full_pii_table())
self.assertEqual({'a': ['PiiTypes.PHONE'], 'b': ['PiiTypes.ADDRESS', 'PiiTypes.LOCATION']}, pii_table)
def test_no_pii(self):
pii_table = GlueStore.get_pii_table(MockExplorer.get_no_pii_table())
self.assertEqual({}, pii_table)
def test_partial_pii(self):
pii_table = GlueStore.get_pii_table(MockExplorer.get_partial_pii_table())
self.assertEqual({'a': ['PiiTypes.PHONE']}, pii_table)
def save_schemas(cls, explorer):
schemas = explorer.get_schemas()
client = boto3.client(
"glue",
region_name=explorer.config.region,
aws_access_key_id=explorer.config.access_key,
aws_secret_access_key=explorer.config.secret_key,
)
logging.debug(client)
for schema in schemas:
logging.debug("Processing schema {0}".format(schema.get_name()))
for table in schema.get_tables():
field_value = GlueStore.get_pii_table(table)
table_info = client.get_table(
DatabaseName=schema.get_name(), Name=table.get_name()
)
logging.debug(table_info)
updated_columns, is_table_updated = GlueStore.update_column_parameters(
table_info["Table"]["StorageDescriptor"]["Columns"], field_value
)
if is_table_updated:
updated_params = GlueStore.update_table_params(
table_info["Table"], updated_columns
)
client.update_table(
DatabaseName=schema.get_name(), TableInput=updated_params