Ada lebih banyak AWS SDK contoh yang tersedia di GitHub repo SDKContoh AWS Dokumen
Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
Contoh Step Functions menggunakan SDK untuk Python (Boto3)
Contoh kode berikut menunjukkan cara melakukan tindakan dan mengimplementasikan skenario umum dengan menggunakan AWS SDK for Python (Boto3) with Step Functions.
Dasar-dasar adalah contoh kode yang menunjukkan kepada Anda bagaimana melakukan operasi penting dalam suatu layanan.
Tindakan adalah kutipan kode dari program yang lebih besar dan harus dijalankan dalam konteks. Sementara tindakan menunjukkan cara memanggil fungsi layanan individual, Anda dapat melihat tindakan dalam konteks dalam skenario terkait.
Skenario adalah contoh kode yang menunjukkan kepada Anda bagaimana menyelesaikan tugas tertentu dengan memanggil beberapa fungsi dalam layanan atau dikombinasikan dengan yang lain Layanan AWS.
Setiap contoh menyertakan tautan ke kode sumber lengkap, di mana Anda dapat menemukan instruksi tentang cara mengatur dan menjalankan kode dalam konteks.
Memulai
Contoh kode berikut menunjukkan cara memulai menggunakan Step Functions.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. import boto3 def hello_stepfunctions(stepfunctions_client): """ Use the AWS SDK for Python (Boto3) to create an AWS Step Functions client and list the state machines in your account. This list might be empty if you haven't created any state machines. This example uses the default settings specified in your shared credentials and config files. :param stepfunctions_client: A Boto3 Step Functions Client object. """ print("Hello, Step Functions! Let's list up to 10 of your state machines:") state_machines = stepfunctions_client.list_state_machines(maxResults=10) for sm in state_machines["stateMachines"]: print(f"\t{sm['name']}: {sm['stateMachineArn']}") if __name__ == "__main__": hello_stepfunctions(boto3.client("stepfunctions"))
-
Untuk API detailnya, lihat ListStateMachines AWSSDKReferensi Python (Boto3). API
-
Hal-hal mendasar
Contoh kode berikut ini menunjukkan cara:
Buat aktivitas.
Buat mesin status dari definisi Amazon States Language yang berisi aktivitas yang dibuat sebelumnya sebagai langkah.
Jalankan mesin status dan tanggapi aktivitas dengan input pengguna.
Dapatkan status dan output akhir setelah proses selesai, lalu bersihkan sumber daya.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. Jalankan skenario interaktif di penggugah/prompt perintah.
class StateMachineScenario: """Runs an interactive scenario that shows how to get started using Step Functions.""" def __init__(self, activity, state_machine, iam_client): """ :param activity: An object that wraps activity actions. :param state_machine: An object that wraps state machine actions. :param iam_client: A Boto3 AWS Identity and Access Management (IAM) client. """ self.activity = activity self.state_machine = state_machine self.iam_client = iam_client self.state_machine_role = None def prerequisites(self, state_machine_role_name): """ Finds or creates an IAM role that can be assumed by Step Functions. A role of this kind is required to create a state machine. The state machine used in this example does not call any additional services, so it needs no additional permissions. :param state_machine_role_name: The name of the role. :return: Data about the role. """ trust_policy = { "Version": "2012-10-17", "Statement": [ { "Sid": "", "Effect": "Allow", "Principal": {"Service": "states.amazonaws.com"}, "Action": "sts:AssumeRole", } ], } try: role = self.iam_client.get_role(RoleName=state_machine_role_name) print(f"Prerequisite IAM role {state_machine_role_name} already exists.") except ClientError as err: if err.response["Error"]["Code"] == "NoSuchEntity": role = None else: logger.error( "Couldn't get prerequisite IAM role %s. Here's why: %s: %s", state_machine_role_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise if role is None: try: role = self.iam_client.create_role( RoleName=state_machine_role_name, AssumeRolePolicyDocument=json.dumps(trust_policy), ) except ClientError as err: logger.error( "Couldn't create prerequisite IAM role %s. Here's why: %s: %s", state_machine_role_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise self.state_machine_role = role["Role"] def find_or_create_activity(self, activity_name): """ Finds or creates a Step Functions activity. :param activity_name: The name of the activity. :return: The Amazon Resource Name (ARN) of the activity. """ print("First, let's set up an activity and state machine.") activity_arn = self.activity.find(activity_name) if activity_arn is None: activity_arn = self.activity.create(activity_name) print( f"Activity {activity_name} created. Its Amazon Resource Name (ARN) is " f"{activity_arn}." ) else: print(f"Activity {activity_name} already exists.") return activity_arn def find_or_create_state_machine( self, state_machine_name, activity_arn, state_machine_file ): """ Finds or creates a Step Functions state machine. :param state_machine_name: The name of the state machine. :param activity_arn: The ARN of an activity that is used as a step in the state machine. This ARN is injected into the state machine definition that's used to create the state machine. :param state_machine_file: The path to a file containing the state machine definition. :return: The ARN of the state machine. """ state_machine_arn = self.state_machine.find(state_machine_name) if state_machine_arn is None: with open(state_machine_file) as state_machine_file: state_machine_def = state_machine_file.read().replace( "{{DOC_EXAMPLE_ACTIVITY_ARN}}", activity_arn ) state_machine_arn = self.state_machine.create( state_machine_name, state_machine_def, self.state_machine_role["Arn"], ) print(f"State machine {state_machine_name} created.") else: print(f"State machine {state_machine_name} already exists.") print("-" * 88) print(f"Here's some information about state machine {state_machine_name}:") state_machine_info = self.state_machine.describe(state_machine_arn) for field in ["name", "status", "stateMachineArn", "roleArn"]: print(f"\t{field}: {state_machine_info[field]}") return state_machine_arn def run_state_machine(self, state_machine_arn, activity_arn): """ Run the state machine. The state machine used in this example is a simple chat simulation. It contains an activity step in a loop that is used for user interaction. When the state machine gets to the activity step, it waits for an external application to get task data and submit a response. This function acts as the activity application by getting task input and responding with user input. :param state_machine_arn: The ARN of the state machine. :param activity_arn: The ARN of the activity used as a step in the state machine. :return: The ARN of the run. """ print( f"Let's run the state machine. It's a simplistic, non-AI chat simulator " f"we'll call ChatSFN." ) user_name = q.ask("What should ChatSFN call you? ", q.non_empty) run_input = {"name": user_name} print("Starting state machine...") run_arn = self.state_machine.start(state_machine_arn, json.dumps(run_input)) action = None while action != "done": activity_task = self.activity.get_task(activity_arn) task_input = json.loads(activity_task["input"]) print(f"ChatSFN: {task_input['message']}") action = task_input["actions"][ q.choose("What now? ", task_input["actions"]) ] task_response = {"action": action} self.activity.send_task_success( activity_task["taskToken"], json.dumps(task_response) ) return run_arn def finish_state_machine_run(self, run_arn): """ Wait for the state machine run to finish, then print final status and output. :param run_arn: The ARN of the run to retrieve. """ print(f"Let's get the final output from the state machine:") status = "RUNNING" while status == "RUNNING": run_output = self.state_machine.describe_run(run_arn) status = run_output["status"] if status == "RUNNING": print( "The state machine is still running, let's wait for it to finish." ) wait(1) elif status == "SUCCEEDED": print(f"ChatSFN: {json.loads(run_output['output'])['message']}") else: print(f"Run status: {status}.") def cleanup( self, state_machine_name, state_machine_arn, activity_name, activity_arn, state_machine_role_name, ): """ Clean up resources created by this example. :param state_machine_name: The name of the state machine. :param state_machine_arn: The ARN of the state machine. :param activity_name: The name of the activity. :param activity_arn: The ARN of the activity. :param state_machine_role_name: The name of the role used by the state machine. """ if q.ask( "Do you want to delete the state machine, activity, and role created for this " "example? (y/n) ", q.is_yesno, ): self.state_machine.delete(state_machine_arn) print(f"Deleted state machine {state_machine_name}.") self.activity.delete(activity_arn) print(f"Deleted activity {activity_name}.") self.iam_client.delete_role(RoleName=state_machine_role_name) print(f"Deleted role {state_machine_role_name}.") def run_scenario(self, activity_name, state_machine_name): print("-" * 88) print("Welcome to the AWS Step Functions state machines demo.") print("-" * 88) activity_arn = self.find_or_create_activity(activity_name) state_machine_arn = self.find_or_create_state_machine( state_machine_name, activity_arn, "../../../resources/sample_files/chat_sfn_state_machine.json", ) print("-" * 88) run_arn = self.run_state_machine(state_machine_arn, activity_arn) print("-" * 88) self.finish_state_machine_run(run_arn) print("-" * 88) self.cleanup( state_machine_name, state_machine_arn, activity_name, activity_arn, self.state_machine_role["RoleName"], ) print("-" * 88) print("\nThanks for watching!") print("-" * 88) if __name__ == "__main__": logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") try: stepfunctions_client = boto3.client("stepfunctions") iam_client = boto3.client("iam") scenario = StateMachineScenario( Activity(stepfunctions_client), StateMachine(stepfunctions_client), iam_client, ) scenario.prerequisites("doc-example-state-machine-chat") scenario.run_scenario("doc-example-activity", "doc-example-state-machine") except Exception: logging.exception("Something went wrong with the demo.")
Tentukan kelas yang membungkus tindakan mesin negara.
class StateMachine: """Encapsulates Step Functions state machine actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def create(self, name, definition, role_arn): """ Creates a state machine with the specific definition. The state machine assumes the provided role before it starts a run. :param name: The name to give the state machine. :param definition: The Amazon States Language definition of the steps in the the state machine. :param role_arn: The Amazon Resource Name (ARN) of the role that is assumed by Step Functions when the state machine is run. :return: The ARN of the newly created state machine. """ try: response = self.stepfunctions_client.create_state_machine( name=name, definition=definition, roleArn=role_arn ) except ClientError as err: logger.error( "Couldn't create state machine %s. Here's why: %s: %s", name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response["stateMachineArn"] def find(self, name): """ Find a state machine by name. This requires listing the state machines until one is found with a matching name. :param name: The name of the state machine to search for. :return: The ARN of the state machine if found; otherwise, None. """ try: paginator = self.stepfunctions_client.get_paginator("list_state_machines") for page in paginator.paginate(): for state_machine in page.get("stateMachines", []): if state_machine["name"] == name: return state_machine["stateMachineArn"] except ClientError as err: logger.error( "Couldn't list state machines. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def describe(self, state_machine_arn): """ Get data about a state machine. :param state_machine_arn: The ARN of the state machine to look up. :return: The retrieved state machine data. """ try: response = self.stepfunctions_client.describe_state_machine( stateMachineArn=state_machine_arn ) except ClientError as err: logger.error( "Couldn't describe state machine %s. Here's why: %s: %s", state_machine_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response def start(self, state_machine_arn, run_input): """ Start a run of a state machine with a specified input. A run is also known as an "execution" in Step Functions. :param state_machine_arn: The ARN of the state machine to run. :param run_input: The input to the state machine, in JSON format. :return: The ARN of the run. This can be used to get information about the run, including its current status and final output. """ try: response = self.stepfunctions_client.start_execution( stateMachineArn=state_machine_arn, input=run_input ) except ClientError as err: logger.error( "Couldn't start state machine %s. Here's why: %s: %s", state_machine_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response["executionArn"] def describe_run(self, run_arn): """ Get data about a state machine run, such as its current status or final output. :param run_arn: The ARN of the run to look up. :return: The retrieved run data. """ try: response = self.stepfunctions_client.describe_execution( executionArn=run_arn ) except ClientError as err: logger.error( "Couldn't describe run %s. Here's why: %s: %s", run_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response def delete(self, state_machine_arn): """ Delete a state machine and all of its run data. :param state_machine_arn: The ARN of the state machine to delete. """ try: response = self.stepfunctions_client.delete_state_machine( stateMachineArn=state_machine_arn ) except ClientError as err: logger.error( "Couldn't delete state machine %s. Here's why: %s: %s", state_machine_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response
Tentukan kelas yang membungkus tindakan aktivitas.
class Activity: """Encapsulates Step Function activity actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def create(self, name): """ Create an activity. :param name: The name of the activity to create. :return: The Amazon Resource Name (ARN) of the newly created activity. """ try: response = self.stepfunctions_client.create_activity(name=name) except ClientError as err: logger.error( "Couldn't create activity %s. Here's why: %s: %s", name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response["activityArn"] def find(self, name): """ Find an activity by name. This requires listing activities until one is found with a matching name. :param name: The name of the activity to search for. :return: If found, the ARN of the activity; otherwise, None. """ try: paginator = self.stepfunctions_client.get_paginator("list_activities") for page in paginator.paginate(): for activity in page.get("activities", []): if activity["name"] == name: return activity["activityArn"] except ClientError as err: logger.error( "Couldn't list activities. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def get_task(self, activity_arn): """ Gets task data for an activity. When a state machine is waiting for the specified activity, a response is returned with data from the state machine. When a state machine is not waiting, this call blocks for 60 seconds. :param activity_arn: The ARN of the activity to get task data for. :return: The task data for the activity. """ try: response = self.stepfunctions_client.get_activity_task( activityArn=activity_arn ) except ClientError as err: logger.error( "Couldn't get a task for activity %s. Here's why: %s: %s", activity_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response def send_task_success(self, task_token, task_response): """ Sends a success response to a waiting activity step. A state machine with an activity step waits for the activity to get task data and then respond with either success or failure before it resumes processing. :param task_token: The token associated with the task. This is included in the response to the get_activity_task action and must be sent without modification. :param task_response: The response data from the activity. This data is received and processed by the state machine. """ try: self.stepfunctions_client.send_task_success( taskToken=task_token, output=task_response ) except ClientError as err: logger.error( "Couldn't send task success. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def delete(self, activity_arn): """ Delete an activity. :param activity_arn: The ARN of the activity to delete. """ try: response = self.stepfunctions_client.delete_activity( activityArn=activity_arn ) except ClientError as err: logger.error( "Couldn't delete activity %s. Here's why: %s: %s", activity_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response
-
Untuk API detailnya, lihat topik berikut AWS SDKuntuk Referensi Python (Boto3). API
-
Tindakan
Contoh kode berikut menunjukkan cara menggunakanCreateActivity
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. class Activity: """Encapsulates Step Function activity actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def create(self, name): """ Create an activity. :param name: The name of the activity to create. :return: The Amazon Resource Name (ARN) of the newly created activity. """ try: response = self.stepfunctions_client.create_activity(name=name) except ClientError as err: logger.error( "Couldn't create activity %s. Here's why: %s: %s", name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response["activityArn"]
-
Untuk API detailnya, lihat CreateActivity AWSSDKReferensi Python (Boto3). API
-
Contoh kode berikut menunjukkan cara menggunakanCreateStateMachine
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. class StateMachine: """Encapsulates Step Functions state machine actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def create(self, name, definition, role_arn): """ Creates a state machine with the specific definition. The state machine assumes the provided role before it starts a run. :param name: The name to give the state machine. :param definition: The Amazon States Language definition of the steps in the the state machine. :param role_arn: The Amazon Resource Name (ARN) of the role that is assumed by Step Functions when the state machine is run. :return: The ARN of the newly created state machine. """ try: response = self.stepfunctions_client.create_state_machine( name=name, definition=definition, roleArn=role_arn ) except ClientError as err: logger.error( "Couldn't create state machine %s. Here's why: %s: %s", name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response["stateMachineArn"]
-
Untuk API detailnya, lihat CreateStateMachine AWSSDKReferensi Python (Boto3). API
-
Contoh kode berikut menunjukkan cara menggunakanDeleteActivity
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. class Activity: """Encapsulates Step Function activity actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def delete(self, activity_arn): """ Delete an activity. :param activity_arn: The ARN of the activity to delete. """ try: response = self.stepfunctions_client.delete_activity( activityArn=activity_arn ) except ClientError as err: logger.error( "Couldn't delete activity %s. Here's why: %s: %s", activity_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response
-
Untuk API detailnya, lihat DeleteActivity AWSSDKReferensi Python (Boto3). API
-
Contoh kode berikut menunjukkan cara menggunakanDeleteStateMachine
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. class StateMachine: """Encapsulates Step Functions state machine actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def delete(self, state_machine_arn): """ Delete a state machine and all of its run data. :param state_machine_arn: The ARN of the state machine to delete. """ try: response = self.stepfunctions_client.delete_state_machine( stateMachineArn=state_machine_arn ) except ClientError as err: logger.error( "Couldn't delete state machine %s. Here's why: %s: %s", state_machine_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response
-
Untuk API detailnya, lihat DeleteStateMachine AWSSDKReferensi Python (Boto3). API
-
Contoh kode berikut menunjukkan cara menggunakanDescribeExecution
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. def describe_run(self, run_arn): """ Get data about a state machine run, such as its current status or final output. :param run_arn: The ARN of the run to look up. :return: The retrieved run data. """ try: response = self.stepfunctions_client.describe_execution( executionArn=run_arn ) except ClientError as err: logger.error( "Couldn't describe run %s. Here's why: %s: %s", run_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response
-
Untuk API detailnya, lihat DescribeExecution AWSSDKReferensi Python (Boto3). API
-
Contoh kode berikut menunjukkan cara menggunakanDescribeStateMachine
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. class StateMachine: """Encapsulates Step Functions state machine actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def describe(self, state_machine_arn): """ Get data about a state machine. :param state_machine_arn: The ARN of the state machine to look up. :return: The retrieved state machine data. """ try: response = self.stepfunctions_client.describe_state_machine( stateMachineArn=state_machine_arn ) except ClientError as err: logger.error( "Couldn't describe state machine %s. Here's why: %s: %s", state_machine_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response
-
Untuk API detailnya, lihat DescribeStateMachine AWSSDKReferensi Python (Boto3). API
-
Contoh kode berikut menunjukkan cara menggunakanGetActivityTask
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. class Activity: """Encapsulates Step Function activity actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def get_task(self, activity_arn): """ Gets task data for an activity. When a state machine is waiting for the specified activity, a response is returned with data from the state machine. When a state machine is not waiting, this call blocks for 60 seconds. :param activity_arn: The ARN of the activity to get task data for. :return: The task data for the activity. """ try: response = self.stepfunctions_client.get_activity_task( activityArn=activity_arn ) except ClientError as err: logger.error( "Couldn't get a task for activity %s. Here's why: %s: %s", activity_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response
-
Untuk API detailnya, lihat GetActivityTask AWSSDKReferensi Python (Boto3). API
-
Contoh kode berikut menunjukkan cara menggunakanListActivities
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. class Activity: """Encapsulates Step Function activity actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def find(self, name): """ Find an activity by name. This requires listing activities until one is found with a matching name. :param name: The name of the activity to search for. :return: If found, the ARN of the activity; otherwise, None. """ try: paginator = self.stepfunctions_client.get_paginator("list_activities") for page in paginator.paginate(): for activity in page.get("activities", []): if activity["name"] == name: return activity["activityArn"] except ClientError as err: logger.error( "Couldn't list activities. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise
-
Untuk API detailnya, lihat ListActivities AWSSDKReferensi Python (Boto3). API
-
Contoh kode berikut menunjukkan cara menggunakanListStateMachines
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. Temukan mesin negara berdasarkan nama dengan mencari daftar mesin negara untuk akun tersebut.
class StateMachine: """Encapsulates Step Functions state machine actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def find(self, name): """ Find a state machine by name. This requires listing the state machines until one is found with a matching name. :param name: The name of the state machine to search for. :return: The ARN of the state machine if found; otherwise, None. """ try: paginator = self.stepfunctions_client.get_paginator("list_state_machines") for page in paginator.paginate(): for state_machine in page.get("stateMachines", []): if state_machine["name"] == name: return state_machine["stateMachineArn"] except ClientError as err: logger.error( "Couldn't list state machines. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise
-
Untuk API detailnya, lihat ListStateMachines AWSSDKReferensi Python (Boto3). API
-
Contoh kode berikut menunjukkan cara menggunakanSendTaskSuccess
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. class Activity: """Encapsulates Step Function activity actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def send_task_success(self, task_token, task_response): """ Sends a success response to a waiting activity step. A state machine with an activity step waits for the activity to get task data and then respond with either success or failure before it resumes processing. :param task_token: The token associated with the task. This is included in the response to the get_activity_task action and must be sent without modification. :param task_response: The response data from the activity. This data is received and processed by the state machine. """ try: self.stepfunctions_client.send_task_success( taskToken=task_token, output=task_response ) except ClientError as err: logger.error( "Couldn't send task success. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise
-
Untuk API detailnya, lihat SendTaskSuccess AWSSDKReferensi Python (Boto3). API
-
Contoh kode berikut menunjukkan cara menggunakanStartExecution
.
- SDKuntuk Python (Boto3)
-
catatan
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara pengaturan dan menjalankannya di Repositori Contoh Kode AWS
. class StateMachine: """Encapsulates Step Functions state machine actions.""" def __init__(self, stepfunctions_client): """ :param stepfunctions_client: A Boto3 Step Functions client. """ self.stepfunctions_client = stepfunctions_client def start(self, state_machine_arn, run_input): """ Start a run of a state machine with a specified input. A run is also known as an "execution" in Step Functions. :param state_machine_arn: The ARN of the state machine to run. :param run_input: The input to the state machine, in JSON format. :return: The ARN of the run. This can be used to get information about the run, including its current status and final output. """ try: response = self.stepfunctions_client.start_execution( stateMachineArn=state_machine_arn, input=run_input ) except ClientError as err: logger.error( "Couldn't start state machine %s. Here's why: %s: %s", state_machine_arn, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response["executionArn"]
-
Untuk API detailnya, lihat StartExecution AWSSDKReferensi Python (Boto3). API
-
Skenario
Contoh kode berikut menunjukkan cara membuat aplikasi AWS Step Functions messenger yang mengambil catatan pesan dari tabel database.
- SDKuntuk Python (Boto3)
-
Menunjukkan cara menggunakan AWS SDK for Python (Boto3) with AWS Step Functions untuk membuat aplikasi messenger yang mengambil catatan pesan dari tabel Amazon DynamoDB dan mengirimkannya dengan Amazon Simple Queue Service (Amazon). SQS Mesin state terintegrasi dengan AWS Lambda fungsi untuk memindai database untuk pesan yang tidak terkirim.
Buat mesin status yang mengambil dan memperbarui catatan pesan dari tabel Amazon DynamoDB.
Perbarui definisi mesin status untuk juga mengirim pesan ke Amazon Simple Queue Service (AmazonSQS).
Mulai dan hentikan berjalannya mesin status.
Connect ke Lambda, DynamoDB, dan SQS Amazon dari mesin status dengan menggunakan integrasi layanan.
Untuk kode sumber lengkap dan instruksi tentang cara mengatur dan menjalankan, lihat contoh lengkapnya di GitHub
. Layanan yang digunakan dalam contoh ini
DynamoDB
Lambda
Amazon SQS
Step Functions
Contoh kode berikut menunjukkan cara membangun dan mengatur aplikasi AI generatif dengan Amazon Bedrock dan Step Functions.
- SDKuntuk Python (Boto3)
-
Skenario Amazon Bedrock Serverless Prompt Chaining menunjukkan bagaimana AWS Step Functions, Amazon Bedrock, https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html dan dapat digunakan untuk membangun dan mengatur aplikasi AI generatif yang kompleks, tanpa server, dan sangat skalabel. Ini berisi contoh kerja berikut:
-
Tulis analisis novel yang diberikan untuk blog sastra. Contoh ini menggambarkan rantai petunjuk yang sederhana dan berurutan.
-
Hasilkan cerita pendek tentang topik tertentu. Contoh ini menggambarkan bagaimana AI dapat secara iteratif memproses daftar item yang dihasilkan sebelumnya.
-
Buat rencana perjalanan untuk liburan akhir pekan ke tujuan tertentu. Contoh ini menggambarkan cara memparalelkan beberapa prompt yang berbeda.
-
Pitch ide film untuk pengguna manusia yang bertindak sebagai produser film. Contoh ini menggambarkan cara memparalelkan prompt yang sama dengan parameter inferensi yang berbeda, cara mundur ke langkah sebelumnya dalam rantai, dan cara memasukkan input manusia sebagai bagian dari alur kerja.
-
Rencanakan makanan berdasarkan bahan-bahan yang dimiliki pengguna. Contoh ini menggambarkan bagaimana rantai cepat dapat menggabungkan dua percakapan AI yang berbeda, dengan dua persona AI terlibat dalam debat satu sama lain untuk meningkatkan hasil akhir.
-
Temukan dan rangkum repositori tren GitHub tertinggi hari ini. Contoh ini menggambarkan rantai beberapa agen AI yang berinteraksi dengan eksternal. APIs
Untuk kode sumber lengkap dan instruksi untuk menyiapkan dan menjalankan, lihat proyek lengkap di GitHub
. Layanan yang digunakan dalam contoh ini
Amazon Bedrock
Runtime Amazon Bedrock
Agen Batuan Dasar Amazon
Runtime Agen Batuan Dasar Amazon
Step Functions
-