How to use get_trace_id method in localstack

Best Python code snippet using localstack_python

organizations.py

Source:organizations.py Github

copy

Full Screen

2from premembers.common import checkauthority, common_utils, eventhelper3from premembers.repository.const import Authority4from premembers.organizations.logic import organizations_logic5def get_organization_handler(event, context):6 trace_id = eventhelper.get_trace_id(event)7 user_id = eventhelper.get_trace_id(event)8 organization_id = eventhelper.get_organization_id(event)9 pm_logger = common_utils.begin_logger(trace_id, __name__,10 inspect.currentframe())11 # アクセス権限チェック12 response_authority = checkauthority.authority(13 trace_id, user_id, organization_id, Authority.Viewer)14 if (response_authority):15 return common_utils.response(response_authority, pm_logger)16 # return data response17 response = organizations_logic.get_organization(trace_id, organization_id)18 return common_utils.response(response, pm_logger)19def create_organization_handler(event, context):20 trace_id = eventhelper.get_trace_id(event)21 email = eventhelper.get_email(event)22 body = event['body']23 pm_logger = common_utils.begin_logger(trace_id, __name__,24 inspect.currentframe())25 # Create data26 response = organizations_logic.create_organization(trace_id, email, body)27 return common_utils.response(response, pm_logger)28def update_organization_handler(event, context):29 trace_id = eventhelper.get_trace_id(event)30 user_id = eventhelper.get_trace_id(event)31 organization_id = eventhelper.get_organization_id(event)32 pm_logger = common_utils.begin_logger(trace_id, __name__,33 inspect.currentframe())34 # アクセス権限チェック35 response_authority = checkauthority.authority(36 trace_id, user_id, organization_id, Authority.Owner)37 if (response_authority):38 return common_utils.response(response_authority, pm_logger)39 # update data40 response = organizations_logic.update_organization(41 trace_id, organization_id, event['body'])42 return common_utils.response(response, pm_logger)43def delete_organization_handler(event, context):44 trace_id = eventhelper.get_trace_id(event)45 user_id = eventhelper.get_trace_id(event)46 email = eventhelper.get_email(event)47 organization_id = eventhelper.get_organization_id(event)48 pm_logger = common_utils.begin_logger(trace_id, __name__,49 inspect.currentframe())50 # アクセス権限チェック51 response_authority = checkauthority.authority(52 trace_id, user_id, organization_id, Authority.Owner)53 if (response_authority):54 return common_utils.response(response_authority, pm_logger)55 # delete data56 response = organizations_logic.delete_organization(57 trace_id, email, organization_id)58 return common_utils.response(response, pm_logger)59def list_users_handler(event, context):60 trace_id = eventhelper.get_trace_id(event)61 user_id = eventhelper.get_trace_id(event)62 organization_id = eventhelper.get_organization_id(event)63 invite_status = eventhelper.get_invite_status(event)64 # Get logging65 pm_logger = common_utils.begin_logger(trace_id, __name__,66 inspect.currentframe())67 # アクセス権限チェック68 response_authority = checkauthority.authority(69 trace_id, user_id, organization_id, Authority.Viewer)70 if (response_authority):71 return common_utils.response(response_authority, pm_logger)72 # return response data73 response = organizations_logic.get_list_users(trace_id, organization_id,74 invite_status)75 return common_utils.response(response, pm_logger)76def delete_user_handler(event, context):77 trace_id = eventhelper.get_trace_id(event)78 user_id_sign_in = eventhelper.get_trace_id(event)79 user_id = eventhelper.get_user_id(event)80 organization_id = eventhelper.get_organization_id(event)81 email = eventhelper.get_email(event)82 # Get logging83 pm_logger = common_utils.begin_logger(trace_id, __name__,84 inspect.currentframe())85 # アクセス権限チェック86 response_authority = checkauthority.authority(87 trace_id, user_id_sign_in, organization_id, Authority.Owner)88 if (response_authority):89 return common_utils.response(response_authority, pm_logger)90 # return response data91 response = organizations_logic.delete_user(trace_id, organization_id,92 user_id, email)93 return common_utils.response(response, pm_logger)94def update_authority_handler(event, context):95 trace_id = eventhelper.get_trace_id(event)96 user_id_sign_in = eventhelper.get_trace_id(event)97 user_id = eventhelper.get_user_id(event)98 organization_id = eventhelper.get_organization_id(event)99 body = event['body']100 # Get logging101 pm_logger = common_utils.begin_logger(trace_id, __name__,102 inspect.currentframe())103 # アクセス権限チェック104 response_authority = checkauthority.authority(105 trace_id, user_id_sign_in, organization_id, Authority.Owner)106 if (response_authority):107 return common_utils.response(response_authority, pm_logger)108 # return response data109 response = organizations_logic.update_authority(trace_id, organization_id,110 user_id, body)111 return common_utils.response(response, pm_logger)112def create_invite_handler(event, context):113 trace_id = eventhelper.get_trace_id(event)114 user_id = eventhelper.get_trace_id(event)115 organization_id = eventhelper.get_organization_id(event)116 body = event['body']117 # Get logging118 pm_logger = common_utils.begin_logger(trace_id, __name__,119 inspect.currentframe())120 # アクセス権限チェック121 response_authority = checkauthority.authority(122 trace_id, user_id, organization_id, Authority.Owner)123 if (response_authority):124 return common_utils.response(response_authority, pm_logger)125 # return response data126 response = organizations_logic.create_invite(127 trace_id, organization_id, body)128 return common_utils.response(response, pm_logger)129def accept_invite_handler(event, context):130 trace_id = eventhelper.get_trace_id(event)131 user_id_sign_in = eventhelper.get_trace_id(event)132 user_id = eventhelper.get_user_id(event)133 organization_id = eventhelper.get_organization_id(event)134 # Get logging135 pm_logger = common_utils.begin_logger(trace_id, __name__,136 inspect.currentframe())137 # return response data138 response = organizations_logic.accept_invite(trace_id, organization_id,139 user_id, user_id_sign_in)140 return common_utils.response(response, pm_logger)141def reject_invite_handler(event, context):142 trace_id = eventhelper.get_trace_id(event)143 user_id_sign_in = eventhelper.get_trace_id(event)144 user_id = eventhelper.get_user_id(event)145 organization_id = eventhelper.get_organization_id(event)146 # Get logging147 pm_logger = common_utils.begin_logger(trace_id, __name__,148 inspect.currentframe())149 # return response data150 response = organizations_logic.reject_invite(trace_id, organization_id,151 user_id, user_id_sign_in)152 return common_utils.response(response, pm_logger)153def execute_force_invites_handler(event, context):154 trace_id = eventhelper.get_trace_id(event)155 user_id = trace_id156 organization_id = eventhelper.get_organization_id(event)157 body = event['body']158 # Get logging159 pm_logger = common_utils.begin_logger(trace_id, __name__,160 inspect.currentframe())161 # アクセス権限チェック162 response_authority = checkauthority.authority(163 trace_id, user_id, organization_id, Authority.Owner)164 if (response_authority):165 return common_utils.response(response_authority, pm_logger)166 # return response data167 response = organizations_logic.execute_force_invites(168 trace_id, body, organization_id)...

Full Screen

Full Screen

awschecks.py

Source:awschecks.py Github

copy

Full Screen

...3from premembers.repository.const import Authority4from premembers.check.logic import awschecks_logic5def get_security_check_summary_handler(event, context):6 # Get data request7 user_id = eventhelper.get_trace_id(event)8 # Get logging9 pm_logger = common_utils.begin_logger(user_id, __name__,10 inspect.currentframe())11 # Get response summary check security12 response = awschecks_logic.get_security_check_summary(user_id)13 return common_utils.response(response, pm_logger)14def get_security_check_detail_handler(event, context):15 trace_id = eventhelper.get_trace_id(event)16 user_id = eventhelper.get_trace_id(event)17 organization_id = eventhelper.get_organization_id(event)18 project_id = eventhelper.get_project_id(event)19 check_history_id = eventhelper.get_check_history_id(event)20 group_filter = eventhelper.get_group_filter(event)21 # Get logging22 pm_logger = common_utils.begin_logger(trace_id, __name__,23 inspect.currentframe())24 # アクセス権限チェックを行います。25 response_authority = checkauthority.authority(26 trace_id, user_id, organization_id, Authority.Viewer)27 if response_authority:28 return common_utils.response(response_authority, pm_logger)29 # return response data30 response = awschecks_logic.get_security_check_detail(31 trace_id, organization_id, project_id, check_history_id, group_filter)32 return common_utils.response(response, pm_logger)33def get_securitycheck_awsacntsummary_handler(event, context):34 # Get data request35 user_id = eventhelper.get_trace_id(event)36 aws_account = eventhelper.get_query_awsaccount(event)37 # Get logging38 pm_logger = common_utils.begin_logger(user_id, __name__,39 inspect.currentframe())40 # return response data41 response = awschecks_logic.get_securitycheck_awsacntsummary(42 user_id, aws_account)43 return common_utils.response(response, pm_logger)44def execute_security_check_handler(event, context):45 trace_id = eventhelper.get_trace_id(event)46 user_id = eventhelper.get_trace_id(event)47 email = eventhelper.get_email(event)48 organization_id = eventhelper.get_organization_id(event)49 project_id = eventhelper.get_project_id(event)50 # Get logging51 pm_logger = common_utils.begin_logger(trace_id, __name__,52 inspect.currentframe())53 # アクセス権限チェックを行います。54 response_authority = checkauthority.authority(55 trace_id, user_id, organization_id, Authority.Editor)56 if response_authority:57 return common_utils.response(response_authority, pm_logger)58 # return response data59 response = awschecks_logic.execute_security_check(trace_id,60 organization_id,61 project_id, user_id,62 email)63 return common_utils.response(response, pm_logger)64def get_security_check_report_url_handler(event, context):65 trace_id = eventhelper.get_trace_id(event)66 user_id = eventhelper.get_trace_id(event)67 history_id = eventhelper.get_history_id(event)68 # Get logging69 pm_logger = common_utils.begin_logger(trace_id, __name__,70 inspect.currentframe())71 # return response data72 response = awschecks_logic.get_security_check_report_url(73 trace_id, user_id, history_id)74 return common_utils.response(response, pm_logger)75def list_security_check_reports_handler(event, context):76 trace_id = eventhelper.get_trace_id(event)77 user_id = eventhelper.get_trace_id(event)78 organization_id = eventhelper.get_organization_id(event)79 project_id = eventhelper.get_project_id(event)80 # Get logging81 pm_logger = common_utils.begin_logger(trace_id, __name__,82 inspect.currentframe())83 # アクセス権限チェックを行います。84 response_authority = checkauthority.authority(85 trace_id, user_id, organization_id, Authority.Viewer)86 if response_authority:87 return common_utils.response(response_authority, pm_logger)88 # return response data89 response = awschecks_logic.list_security_check_reports(90 trace_id, organization_id, project_id)91 return common_utils.response(response, pm_logger)92def get_security_check_webhook_handler(event, context):93 trace_id = eventhelper.get_trace_id(event)94 user_id = eventhelper.get_trace_id(event)95 organization_id = eventhelper.get_query_organization_id(event)96 project_id = eventhelper.get_query_project_id(event)97 pm_logger = common_utils.begin_logger(trace_id, __name__,98 inspect.currentframe())99 # アクセス権限チェックを行います。100 response_authority = checkauthority.authority(101 trace_id, user_id, organization_id, Authority.Editor)102 if response_authority:103 return common_utils.response(response_authority, pm_logger)104 response = awschecks_logic.get_security_check_webhook_by_ids(105 trace_id, user_id, organization_id, project_id)106 return common_utils.response(response, pm_logger)107def generate_security_check_webhook_handler(event, context):108 trace_id = eventhelper.get_trace_id(event)109 user_id = eventhelper.get_trace_id(event)110 email = eventhelper.get_email(event)111 body = eventhelper.parse_body(event)112 organization_id = body['organizationId']113 project_id = body['projectId']114 pm_logger = common_utils.begin_logger(trace_id, __name__,115 inspect.currentframe())116 # アクセス権限チェックを行います。117 response_authority = checkauthority.authority(118 trace_id, user_id, organization_id, Authority.Editor)119 if response_authority:120 return common_utils.response(response_authority, pm_logger)121 response = awschecks_logic.generate_security_check_webhook(122 trace_id, organization_id, project_id, user_id, email)123 return common_utils.response(response, pm_logger)124def update_security_check_webhook_handler(event, context):125 trace_id = eventhelper.get_trace_id(event)126 user_id = eventhelper.get_trace_id(event)127 webhook_path = eventhelper.get_webhook_path(event)128 body = eventhelper.parse_body(event)129 pm_logger = common_utils.begin_logger(trace_id, __name__,130 inspect.currentframe())131 response = awschecks_logic.update_security_check_webhook(132 trace_id, webhook_path, user_id, body)133 return common_utils.response(response, pm_logger)134def execute_security_check_webhook_handler(event, context):135 webhook_path = eventhelper.get_webhook_path(event)136 trace_id = webhook_path137 # Get logging138 pm_logger = common_utils.begin_logger(trace_id, __name__,139 inspect.currentframe())140 response = awschecks_logic.execute_security_check_from_external(141 trace_id, webhook_path)142 return common_utils.response(response, pm_logger)143def get_security_check_resource_handler(event, context):144 trace_id = eventhelper.get_trace_id(event)145 user_id = trace_id146 organization_id = eventhelper.get_organization_id(event)147 coop_id = eventhelper.get_coop_id(event)148 project_id = eventhelper.get_project_id(event)149 check_item_code = eventhelper.get_check_item_code(event)150 # Get logging151 pm_logger = common_utils.begin_logger(trace_id, __name__,152 inspect.currentframe())153 # アクセス権限チェックを行います。154 response_authority = checkauthority.authority(155 trace_id, user_id, organization_id, Authority.Viewer)156 if response_authority:157 return common_utils.response(response_authority, pm_logger)158 # return response data...

Full Screen

Full Screen

lambda_utils.py

Source:lambda_utils.py Github

copy

Full Screen

...9 def __init__(self):10 self.logger = Logger()11 self.tracer = Tracer()12 def set_trace_id(self, lambda_handler):13 def get_trace_id(raw_trace_id):14 for t in raw_trace_id.split(';'):15 if t.startswith('Root'):16 return t.split('=')[1]17 return None18 def decorate(event, context):19 if '_X_AMZN_TRACE_ID' in os.environ:20 if get_trace_id(os.environ['_X_AMZN_TRACE_ID']) is not None:21 os.environ['trace_id'] = get_trace_id(22 os.environ['_X_AMZN_TRACE_ID'])23 if 'trace_id' not in os.environ:24 START_TIME = time.time()25 HEX = hex(int(START_TIME))[2:]26 os.environ['trace_id'] = "0-{}-{}".format(27 HEX, str(binascii.hexlify(urandom(12)), 'utf-8'))28 self.logger.structure_logs(29 append=True, trace_id=os.environ['trace_id'])30 self.logger.info(os.environ.get('trace_id'))31 response = lambda_handler(event, context)32 return response33 return decorate34 def api_gateway_response(self, lambda_handler):35 def decorate(event, context):...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful