Ada lebih banyak contoh AWS SDK yang tersedia di repo Contoh [SDK AWS Doc](https://github.com/awsdocs/aws-doc-sdk-examples). GitHub Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris. # Contoh Amazon Keyspaces menggunakan SDK for Python (Boto3) Contoh kode berikut menunjukkan cara melakukan tindakan dan mengimplementasikan skenario umum dengan menggunakan AWS SDK untuk Python (Boto3) with Amazon Keyspaces. *Dasar-dasar* adalah contoh kode yang menunjukkan kepada Anda bagaimana melakukan operasi penting dalam suatu layanan. *Tindakan* merupakan 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. Setiap contoh menyertakan tautan ke kode sumber lengkap, di mana Anda dapat menemukan instruksi tentang cara mengatur dan menjalankan kode dalam konteks. **Topics** + [Memulai](#get_started) + [Hal-hal mendasar](#basics) + [Tindakan](#actions) ## Memulai ### Halo Amazon Keyspaces Contoh kode berikut menunjukkan cara memulai menggunakan Amazon Keyspaces. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` import boto3 def hello_keyspaces(keyspaces_client): """ Use the AWS SDK for Python (Boto3) to create an Amazon Keyspaces (for Apache Cassandra) client and list the keyspaces in your account. This example uses the default settings specified in your shared credentials and config files. :param keyspaces_client: A Boto3 Amazon Keyspaces Client object. This object wraps the low-level Amazon Keyspaces service API. """ print("Hello, Amazon Keyspaces! Let's list some of your keyspaces:\n") for ks in keyspaces_client.list_keyspaces(maxResults=5).get("keyspaces", []): print(ks["keyspaceName"]) print(f"\t{ks['resourceArn']}") if __name__ == "__main__": hello_keyspaces(boto3.client("keyspaces")) ``` + Untuk detail API, lihat [ListKeyspaces](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/ListKeyspaces)di *AWS SDK for Python (Boto3) Referensi* API. ## Hal-hal mendasar ### Pelajari dasar-dasarnya Contoh kode berikut ini menunjukkan cara untuk melakukan: + Buat keyspace dan tabel. Skema tabel menyimpan data film dan mengaktifkan point-in-time pemulihan. + Connect ke keyspace menggunakan koneksi TLS aman dengan otentikasi SiGv4. + Kueri tabel. Tambahkan, ambil, dan perbarui data film. + Perbarui tabel. Tambahkan kolom untuk melacak film yang ditonton. + Kembalikan tabel ke keadaan sebelumnya dan bersihkan sumber daya. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkap dan pelajari cara menyiapkan dan menjalankan di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). Jalankan skenario interaktif di penggugah/prompt perintah. ``` class KeyspaceScenario: """Runs an interactive scenario that shows how to get started using Amazon Keyspaces.""" def __init__(self, ks_wrapper): """ :param ks_wrapper: An object that wraps Amazon Keyspace actions. """ self.ks_wrapper = ks_wrapper @demo_func def create_keyspace(self): """ 1. Creates a keyspace. 2. Lists up to 10 keyspaces in your account. """ print("Let's create a keyspace.") ks_name = q.ask( "Enter a name for your new keyspace.\nThe name can contain only letters, " "numbers and underscores: ", q.non_empty, ) if self.ks_wrapper.exists_keyspace(ks_name): print(f"A keyspace named {ks_name} exists.") else: ks_arn = self.ks_wrapper.create_keyspace(ks_name) ks_exists = False while not ks_exists: wait(3) ks_exists = self.ks_wrapper.exists_keyspace(ks_name) print(f"Created a new keyspace.\n\t{ks_arn}.") print("The first 10 keyspaces in your account are:\n") self.ks_wrapper.list_keyspaces(10) @demo_func def create_table(self): """ 1. Creates a table in the keyspace. The table is configured with a schema to hold movie data and has point-in-time recovery enabled. 2. Waits for the table to be in an active state. 3. Displays schema information for the table. 4. Lists tables in the keyspace. """ print("Let's create a table for movies in your keyspace.") table_name = q.ask("Enter a name for your table: ", q.non_empty) table = self.ks_wrapper.get_table(table_name) if table is not None: print( f"A table named {table_name} already exists in keyspace " f"{self.ks_wrapper.ks_name}." ) else: table_arn = self.ks_wrapper.create_table(table_name) print(f"Created table {table_name}:\n\t{table_arn}") table = {"status": None} print("Waiting for your table to be ready...") while table["status"] != "ACTIVE": wait(5) table = self.ks_wrapper.get_table(table_name) print(f"Your table is {table['status']}. Its schema is:") pp(table["schemaDefinition"]) print("\nThe tables in your keyspace are:\n") self.ks_wrapper.list_tables() @demo_func def ensure_tls_cert(self): """ Ensures you have a TLS certificate available to use to secure the connection to the keyspace. This function downloads a default certificate or lets you specify your own. """ print("To connect to your keyspace, you must have a TLS certificate.") print("Checking for TLS certificate...") cert_path = os.path.join( os.path.dirname(__file__), QueryManager.DEFAULT_CERT_FILE ) if not os.path.exists(cert_path): cert_choice = q.ask( f"Press enter to download a certificate from {QueryManager.CERT_URL} " f"or enter the full path to the certificate you want to use: " ) if cert_choice: cert_path = cert_choice else: cert = requests.get(QueryManager.CERT_URL).text with open(cert_path, "w") as cert_file: cert_file.write(cert) else: q.ask(f"Certificate {cert_path} found. Press Enter to continue.") print( f"Certificate {cert_path} will be used to secure the connection to your keyspace." ) return cert_path @demo_func def query_table(self, qm, movie_file): """ 1. Adds movies to the table from a sample movie data file. 2. Gets a list of movies from the table and lets you select one. 3. Displays more information about the selected movie. """ qm.add_movies(self.ks_wrapper.table_name, movie_file) movies = qm.get_movies(self.ks_wrapper.table_name) print(f"Added {len(movies)} movies to the table:") sel = q.choose("Pick one to learn more about it: ", [m.title for m in movies]) movie_choice = qm.get_movie( self.ks_wrapper.table_name, movies[sel].title, movies[sel].year ) print(movie_choice.title) print(f"\tReleased: {movie_choice.release_date}") print(f"\tPlot: {movie_choice.plot}") @demo_func def update_and_restore_table(self, qm): """ 1. Updates the table by adding a column to track watched movies. 2. Marks some of the movies as watched. 3. Gets the list of watched movies from the table. 4. Restores to a movies_restored table at a previous point in time. 5. Gets the list of movies from the restored table. """ print("Let's add a column to record which movies you've watched.") pre_update_timestamp = datetime.utcnow() print( f"Recorded the current UTC time of {pre_update_timestamp} so we can restore the table later." ) self.ks_wrapper.update_table() print("Waiting for your table to update...") table = {"status": "UPDATING"} while table["status"] != "ACTIVE": wait(5) table = self.ks_wrapper.get_table(self.ks_wrapper.table_name) print("Column 'watched' added to table.") q.ask( "Let's mark some of the movies as watched. Press Enter when you're ready.\n" ) movies = qm.get_movies(self.ks_wrapper.table_name) for movie in movies[:10]: qm.watched_movie(self.ks_wrapper.table_name, movie.title, movie.year) print(f"Marked {movie.title} as watched.") movies = qm.get_movies(self.ks_wrapper.table_name, watched=True) print("-" * 88) print("The watched movies in our table are:\n") for movie in movies: print(movie.title) print("-" * 88) if q.ask( "Do you want to restore the table to the way it was before all of these\n" "updates? Keep in mind, this can take up to 20 minutes. (y/n) ", q.is_yesno, ): starting_table_name = self.ks_wrapper.table_name table_name_restored = self.ks_wrapper.restore_table(pre_update_timestamp) table = {"status": "RESTORING"} while table["status"] != "ACTIVE": wait(10) table = self.ks_wrapper.get_table(table_name_restored) print( f"Restored {starting_table_name} to {table_name_restored} " f"at a point in time of {pre_update_timestamp}." ) movies = qm.get_movies(table_name_restored) print("Now the movies in our table are:") for movie in movies: print(movie.title) def cleanup(self, cert_path): """ 1. Deletes the table and waits for it to be removed. 2. Deletes the keyspace. :param cert_path: The path of the TLS certificate used in the demo. If the certificate was downloaded during the demo, it is removed. """ if q.ask( f"Do you want to delete your {self.ks_wrapper.table_name} table and " f"{self.ks_wrapper.ks_name} keyspace? (y/n) ", q.is_yesno, ): table_name = self.ks_wrapper.table_name self.ks_wrapper.delete_table() table = self.ks_wrapper.get_table(table_name) print("Waiting for the table to be deleted.") while table is not None: wait(5) table = self.ks_wrapper.get_table(table_name) print("Table deleted.") self.ks_wrapper.delete_keyspace() print( "Keyspace deleted. If you chose to restore your table during the " "demo, the original table is also deleted." ) if cert_path == os.path.join( os.path.dirname(__file__), QueryManager.DEFAULT_CERT_FILE ) and os.path.exists(cert_path): os.remove(cert_path) print("Removed certificate that was downloaded for this demo.") def run_scenario(self): logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") print("-" * 88) print("Welcome to the Amazon Keyspaces (for Apache Cassandra) demo.") print("-" * 88) self.create_keyspace() self.create_table() cert_file_path = self.ensure_tls_cert() # Use a context manager to ensure the connection to the keyspace is closed. with QueryManager( cert_file_path, boto3.DEFAULT_SESSION, self.ks_wrapper.ks_name ) as qm: self.query_table(qm, "../../../resources/sample_files/movies.json") self.update_and_restore_table(qm) self.cleanup(cert_file_path) print("\nThanks for watching!") print("-" * 88) if __name__ == "__main__": try: scenario = KeyspaceScenario(KeyspaceWrapper.from_client()) scenario.run_scenario() except Exception: logging.exception("Something went wrong with the demo.") ``` Tentukan kelas yang membungkus tindakan keyspace dan tabel. ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def create_keyspace(self, name): """ Creates a keyspace. :param name: The name to give the keyspace. :return: The Amazon Resource Name (ARN) of the new keyspace. """ try: response = self.keyspaces_client.create_keyspace(keyspaceName=name) self.ks_name = name self.ks_arn = response["resourceArn"] except ClientError as err: logger.error( "Couldn't create %s. Here's why: %s: %s", name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return self.ks_arn def exists_keyspace(self, name): """ Checks whether a keyspace exists. :param name: The name of the keyspace to look up. :return: True when the keyspace exists. Otherwise, False. """ try: response = self.keyspaces_client.get_keyspace(keyspaceName=name) self.ks_name = response["keyspaceName"] self.ks_arn = response["resourceArn"] exists = True except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": logger.info("Keyspace %s does not exist.", name) exists = False else: logger.error( "Couldn't verify %s exists. Here's why: %s: %s", name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return exists def list_keyspaces(self, limit): """ Lists the keyspaces in your account. :param limit: The maximum number of keyspaces to list. """ try: ks_paginator = self.keyspaces_client.get_paginator("list_keyspaces") for page in ks_paginator.paginate(PaginationConfig={"MaxItems": limit}): for ks in page["keyspaces"]: print(ks["keyspaceName"]) print(f"\t{ks['resourceArn']}") except ClientError as err: logger.error( "Couldn't list keyspaces. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def create_table(self, table_name): """ Creates a table in the keyspace. The table is created with a schema for storing movie data and has point-in-time recovery enabled. :param table_name: The name to give the table. :return: The ARN of the new table. """ try: response = self.keyspaces_client.create_table( keyspaceName=self.ks_name, tableName=table_name, schemaDefinition={ "allColumns": [ {"name": "title", "type": "text"}, {"name": "year", "type": "int"}, {"name": "release_date", "type": "timestamp"}, {"name": "plot", "type": "text"}, ], "partitionKeys": [{"name": "year"}, {"name": "title"}], }, pointInTimeRecovery={"status": "ENABLED"}, ) except ClientError as err: logger.error( "Couldn't create table %s. Here's why: %s: %s", table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response["resourceArn"] def get_table(self, table_name): """ Gets data about a table in the keyspace. :param table_name: The name of the table to look up. :return: Data about the table. """ try: response = self.keyspaces_client.get_table( keyspaceName=self.ks_name, tableName=table_name ) self.table_name = table_name except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": logger.info("Table %s does not exist.", table_name) self.table_name = None response = None else: logger.error( "Couldn't verify %s exists. Here's why: %s: %s", table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return response def list_tables(self): """ Lists the tables in the keyspace. """ try: table_paginator = self.keyspaces_client.get_paginator("list_tables") for page in table_paginator.paginate(keyspaceName=self.ks_name): for table in page["tables"]: print(table["tableName"]) print(f"\t{table['resourceArn']}") except ClientError as err: logger.error( "Couldn't list tables in keyspace %s. Here's why: %s: %s", self.ks_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def update_table(self): """ Updates the schema of the table. This example updates a table of movie data by adding a new column that tracks whether the movie has been watched. """ try: self.keyspaces_client.update_table( keyspaceName=self.ks_name, tableName=self.table_name, addColumns=[{"name": "watched", "type": "boolean"}], ) except ClientError as err: logger.error( "Couldn't update table %s. Here's why: %s: %s", self.table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def restore_table(self, restore_timestamp): """ Restores the table to a previous point in time. The table is restored to a new table in the same keyspace. :param restore_timestamp: The point in time to restore the table. This time must be in UTC format. :return: The name of the restored table. """ try: restored_table_name = f"{self.table_name}_restored" self.keyspaces_client.restore_table( sourceKeyspaceName=self.ks_name, sourceTableName=self.table_name, targetKeyspaceName=self.ks_name, targetTableName=restored_table_name, restoreTimestamp=restore_timestamp, ) except ClientError as err: logger.error( "Couldn't restore table %s. Here's why: %s: %s", restore_timestamp, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return restored_table_name def delete_table(self): """ Deletes the table from the keyspace. """ try: self.keyspaces_client.delete_table( keyspaceName=self.ks_name, tableName=self.table_name ) self.table_name = None except ClientError as err: logger.error( "Couldn't delete table %s. Here's why: %s: %s", self.table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise def delete_keyspace(self): """ Deletes the keyspace. """ try: self.keyspaces_client.delete_keyspace(keyspaceName=self.ks_name) self.ks_name = None except ClientError as err: logger.error( "Couldn't delete keyspace %s. Here's why: %s: %s", self.ks_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise ``` Tentukan kelas yang membuat koneksi TLS ke ruang kunci, mengautentikasi dengan SigV4, dan mengirimkan kueri CQL ke tabel di ruang kunci. ``` class QueryManager: """ Manages queries to an Amazon Keyspaces (for Apache Cassandra) keyspace. Queries are secured by TLS and authenticated by using the Signature V4 (SigV4) AWS signing protocol. This is more secure than sending username and password with a plain-text authentication provider. This example downloads a default certificate to secure TLS, or lets you specify your own. This example uses a table of movie data to demonstrate basic queries. """ DEFAULT_CERT_FILE = "sf-class2-root.crt" CERT_URL = f"https://certs.secureserver.net/repository/sf-class2-root.crt" def __init__(self, cert_file_path, boto_session, keyspace_name): """ :param cert_file_path: The path and file name of the certificate used for TLS. :param boto_session: A Boto3 session. This is used to acquire your AWS credentials. :param keyspace_name: The name of the keyspace to connect. """ self.cert_file_path = cert_file_path self.boto_session = boto_session self.ks_name = keyspace_name self.cluster = None self.session = None def __enter__(self): """ Creates a session connection to the keyspace that is secured by TLS and authenticated by SigV4. """ ssl_context = SSLContext(PROTOCOL_TLSv1_2) ssl_context.load_verify_locations(self.cert_file_path) ssl_context.verify_mode = CERT_REQUIRED auth_provider = SigV4AuthProvider(self.boto_session) contact_point = f"cassandra.{self.boto_session.region_name}.amazonaws.com" exec_profile = ExecutionProfile( consistency_level=ConsistencyLevel.LOCAL_QUORUM, load_balancing_policy=DCAwareRoundRobinPolicy(), ) self.cluster = Cluster( [contact_point], ssl_context=ssl_context, auth_provider=auth_provider, port=9142, execution_profiles={EXEC_PROFILE_DEFAULT: exec_profile}, protocol_version=4, ) self.cluster.__enter__() self.session = self.cluster.connect(self.ks_name) return self def __exit__(self, *args): """ Exits the cluster. This shuts down all existing session connections. """ self.cluster.__exit__(*args) def add_movies(self, table_name, movie_file_path): """ Gets movies from a JSON file and adds them to a table in the keyspace. :param table_name: The name of the table. :param movie_file_path: The path and file name of a JSON file that contains movie data. """ with open(movie_file_path, "r") as movie_file: movies = json.loads(movie_file.read()) stmt = self.session.prepare( f"INSERT INTO {table_name} (year, title, release_date, plot) VALUES (?, ?, ?, ?);" ) for movie in movies[:20]: self.session.execute( stmt, parameters=[ movie["year"], movie["title"], date.fromisoformat(movie["info"]["release_date"].partition("T")[0]), movie["info"]["plot"], ], ) def get_movies(self, table_name, watched=None): """ Gets the title and year of the full list of movies from the table. :param table_name: The name of the movie table. :param watched: When specified, the returned list of movies is filtered to either movies that have been watched or movies that have not been watched. Otherwise, all movies are returned. :return: A list of movies in the table. """ if watched is None: stmt = SimpleStatement(f"SELECT title, year from {table_name}") params = None else: stmt = SimpleStatement( f"SELECT title, year from {table_name} WHERE watched = %s ALLOW FILTERING" ) params = [watched] return self.session.execute(stmt, parameters=params).all() def get_movie(self, table_name, title, year): """ Gets a single movie from the table, by title and year. :param table_name: The name of the movie table. :param title: The title of the movie. :param year: The year of the movie's release. :return: The requested movie. """ return self.session.execute( SimpleStatement( f"SELECT * from {table_name} WHERE title = %s AND year = %s" ), parameters=[title, year], ).one() def watched_movie(self, table_name, title, year): """ Updates a movie as having been watched. :param table_name: The name of the movie table. :param title: The title of the movie. :param year: The year of the movie's release. """ self.session.execute( SimpleStatement( f"UPDATE {table_name} SET watched=true WHERE title = %s AND year = %s" ), parameters=[title, year], ) ``` + Untuk detail API, lihat topik berikut di *Referensi API AWS SDK untuk Python (Boto3)*. + [CreateKeyspace](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/CreateKeyspace) + [CreateTable](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/CreateTable) + [DeleteKeyspace](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/DeleteKeyspace) + [DeleteTable](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/DeleteTable) + [GetKeyspace](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/GetKeyspace) + [GetTable](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/GetTable) + [ListKeyspaces](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/ListKeyspaces) + [ListTables](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/ListTables) + [RestoreTable](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/RestoreTable) + [UpdateTable](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/UpdateTable) ## Tindakan ### `CreateKeyspace` Contoh kode berikut menunjukkan cara menggunakan`CreateKeyspace`. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def create_keyspace(self, name): """ Creates a keyspace. :param name: The name to give the keyspace. :return: The Amazon Resource Name (ARN) of the new keyspace. """ try: response = self.keyspaces_client.create_keyspace(keyspaceName=name) self.ks_name = name self.ks_arn = response["resourceArn"] except ClientError as err: logger.error( "Couldn't create %s. Here's why: %s: %s", name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return self.ks_arn ``` + Untuk detail API, lihat [CreateKeyspace](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/CreateKeyspace)di *AWS SDK for Python (Boto3) Referensi* API. ### `CreateTable` Contoh kode berikut menunjukkan cara menggunakan`CreateTable`. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def create_table(self, table_name): """ Creates a table in the keyspace. The table is created with a schema for storing movie data and has point-in-time recovery enabled. :param table_name: The name to give the table. :return: The ARN of the new table. """ try: response = self.keyspaces_client.create_table( keyspaceName=self.ks_name, tableName=table_name, schemaDefinition={ "allColumns": [ {"name": "title", "type": "text"}, {"name": "year", "type": "int"}, {"name": "release_date", "type": "timestamp"}, {"name": "plot", "type": "text"}, ], "partitionKeys": [{"name": "year"}, {"name": "title"}], }, pointInTimeRecovery={"status": "ENABLED"}, ) except ClientError as err: logger.error( "Couldn't create table %s. Here's why: %s: %s", table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return response["resourceArn"] ``` + Untuk detail API, lihat [CreateTable](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/CreateTable)di *AWS SDK for Python (Boto3) Referensi* API. ### `DeleteKeyspace` Contoh kode berikut menunjukkan cara menggunakan`DeleteKeyspace`. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def delete_keyspace(self): """ Deletes the keyspace. """ try: self.keyspaces_client.delete_keyspace(keyspaceName=self.ks_name) self.ks_name = None except ClientError as err: logger.error( "Couldn't delete keyspace %s. Here's why: %s: %s", self.ks_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise ``` + Untuk detail API, lihat [DeleteKeyspace](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/DeleteKeyspace)di *AWS SDK for Python (Boto3) Referensi* API. ### `DeleteTable` Contoh kode berikut menunjukkan cara menggunakan`DeleteTable`. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def delete_table(self): """ Deletes the table from the keyspace. """ try: self.keyspaces_client.delete_table( keyspaceName=self.ks_name, tableName=self.table_name ) self.table_name = None except ClientError as err: logger.error( "Couldn't delete table %s. Here's why: %s: %s", self.table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise ``` + Untuk detail API, lihat [DeleteTable](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/DeleteTable)di *AWS SDK for Python (Boto3) Referensi* API. ### `GetKeyspace` Contoh kode berikut menunjukkan cara menggunakan`GetKeyspace`. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def exists_keyspace(self, name): """ Checks whether a keyspace exists. :param name: The name of the keyspace to look up. :return: True when the keyspace exists. Otherwise, False. """ try: response = self.keyspaces_client.get_keyspace(keyspaceName=name) self.ks_name = response["keyspaceName"] self.ks_arn = response["resourceArn"] exists = True except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": logger.info("Keyspace %s does not exist.", name) exists = False else: logger.error( "Couldn't verify %s exists. Here's why: %s: %s", name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return exists ``` + Untuk detail API, lihat [GetKeyspace](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/GetKeyspace)di *AWS SDK for Python (Boto3) Referensi* API. ### `GetTable` Contoh kode berikut menunjukkan cara menggunakan`GetTable`. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def get_table(self, table_name): """ Gets data about a table in the keyspace. :param table_name: The name of the table to look up. :return: Data about the table. """ try: response = self.keyspaces_client.get_table( keyspaceName=self.ks_name, tableName=table_name ) self.table_name = table_name except ClientError as err: if err.response["Error"]["Code"] == "ResourceNotFoundException": logger.info("Table %s does not exist.", table_name) self.table_name = None response = None else: logger.error( "Couldn't verify %s exists. Here's why: %s: %s", table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise return response ``` + Untuk detail API, lihat [GetTable](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/GetTable)di *AWS SDK for Python (Boto3) Referensi* API. ### `ListKeyspaces` Contoh kode berikut menunjukkan cara menggunakan`ListKeyspaces`. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def list_keyspaces(self, limit): """ Lists the keyspaces in your account. :param limit: The maximum number of keyspaces to list. """ try: ks_paginator = self.keyspaces_client.get_paginator("list_keyspaces") for page in ks_paginator.paginate(PaginationConfig={"MaxItems": limit}): for ks in page["keyspaces"]: print(ks["keyspaceName"]) print(f"\t{ks['resourceArn']}") except ClientError as err: logger.error( "Couldn't list keyspaces. Here's why: %s: %s", err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise ``` + Untuk detail API, lihat [ListKeyspaces](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/ListKeyspaces)di *AWS SDK for Python (Boto3) Referensi* API. ### `ListTables` Contoh kode berikut menunjukkan cara menggunakan`ListTables`. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def list_tables(self): """ Lists the tables in the keyspace. """ try: table_paginator = self.keyspaces_client.get_paginator("list_tables") for page in table_paginator.paginate(keyspaceName=self.ks_name): for table in page["tables"]: print(table["tableName"]) print(f"\t{table['resourceArn']}") except ClientError as err: logger.error( "Couldn't list tables in keyspace %s. Here's why: %s: %s", self.ks_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise ``` + Untuk detail API, lihat [ListTables](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/ListTables)di *AWS SDK for Python (Boto3) Referensi* API. ### `RestoreTable` Contoh kode berikut menunjukkan cara menggunakan`RestoreTable`. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def restore_table(self, restore_timestamp): """ Restores the table to a previous point in time. The table is restored to a new table in the same keyspace. :param restore_timestamp: The point in time to restore the table. This time must be in UTC format. :return: The name of the restored table. """ try: restored_table_name = f"{self.table_name}_restored" self.keyspaces_client.restore_table( sourceKeyspaceName=self.ks_name, sourceTableName=self.table_name, targetKeyspaceName=self.ks_name, targetTableName=restored_table_name, restoreTimestamp=restore_timestamp, ) except ClientError as err: logger.error( "Couldn't restore table %s. Here's why: %s: %s", restore_timestamp, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise else: return restored_table_name ``` + Untuk detail API, lihat [RestoreTable](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/RestoreTable)di *AWS SDK for Python (Boto3) Referensi* API. ### `UpdateTable` Contoh kode berikut menunjukkan cara menggunakan`UpdateTable`. **SDK untuk Python (Boto3)** Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/keyspaces#code-examples). ``` class KeyspaceWrapper: """Encapsulates Amazon Keyspaces (for Apache Cassandra) keyspace and table actions.""" def __init__(self, keyspaces_client): """ :param keyspaces_client: A Boto3 Amazon Keyspaces client. """ self.keyspaces_client = keyspaces_client self.ks_name = None self.ks_arn = None self.table_name = None @classmethod def from_client(cls): keyspaces_client = boto3.client("keyspaces") return cls(keyspaces_client) def update_table(self): """ Updates the schema of the table. This example updates a table of movie data by adding a new column that tracks whether the movie has been watched. """ try: self.keyspaces_client.update_table( keyspaceName=self.ks_name, tableName=self.table_name, addColumns=[{"name": "watched", "type": "boolean"}], ) except ClientError as err: logger.error( "Couldn't update table %s. Here's why: %s: %s", self.table_name, err.response["Error"]["Code"], err.response["Error"]["Message"], ) raise ``` + Untuk detail API, lihat [UpdateTable](https://docs.aws.amazon.com/goto/boto3/keyspaces-2022-02-10/UpdateTable)di *AWS SDK for Python (Boto3) Referensi* API.