SDK für Python mit Streaming verwenden - Schnellstart

1. Zur Verwendung des SDK für Python ist Folgendes erforderlich:

   • Ein Oracle Cloud Infrastructure-Account.
   • Ein in diesem Account erstellter Benutzer in einer Gruppe mit einer Policy, die die erforderlichen Berechtigungen erteilt. Dieser Benutzer kann Ihr eigener Benutzer oder eine andere Person/ein anderes System sein, die/das die API aufrufen muss. Ein Beispiel für die Einrichtung eines neuen Benutzers, einer neuen Gruppe, eines neuen Compartments und einer neuen Policy finden Sie unter Benutzer hinzufügen. Eine Liste der typischen Policys, die Sie verwenden können, finden Sie unter Allgemeine Policys.
   • Ein Schlüsselpaar zum Signieren von API-Anforderungen, wobei der Public Key bei Oracle hochgeladen wird. Nur der Benutzer, der die API aufruft, sollte im Besitz des Private Keys sein.
   Hinweis
   
   Weitere Informationen finden Sie unter SDK konfigurieren.
2. Erfassen Sie den Nachrichtenendpunkt und die OCID eines Streams. Die Schritte zum Abrufen von Details für einen Stream finden Sie unter Details für einen Stream abrufen. Im Rahmen dieses Schnellstarts sollte der Stream einen öffentlichen Endpunkt und die von Oracle verwaltete Verschlüsselung verwenden. Informationen hierzu finden Sie unter Streams erstellen und Streampool erstellen, wenn kein Stream vorhanden ist.
3. Python 3.6 oder höher, mit installiertem und aktualisiertem PIP.
4. Visual Code Studio (empfohlen) oder eine andere Integrated Development Environment (IDE).

5. Installieren Sie oci-sdk-Packages für Python mit dem folgenden Befehl:

   pip install oci

   Hinweis
   
   Es wird empfohlen, bei der Installation von oci eine virtuelle Python-Umgebung zu verwenden. Weitere Informationen finden Sie unter SDK herunterladen und installieren.
6. Stellen Sie sicher, dass Sie über eine gültige SDK-Konfigurationsdatei verfügen. Verwenden Sie für Produktionsumgebungen die Instanz-Principal-Autorisierung.

1. Öffnen Sie Ihren bevorzugten Editor, wie Visual Studio Code, im Verzeichnis wd. oci-sdk-Packages für Python sollten bereits in der aktuellen Python-Umgebung installiert sein, nachdem Sie die Schritte unter Voraussetzungen ausgeführt haben.

2. Erstellen Sie eine Datei namens Producer.py im Verzeichnis wd mit dem folgenden Code. Ersetzen Sie die Werte der Variablen ociConfigFilePath, ociProfileName,ociStreamOcid und ociMessageEndpoint im folgenden Code-Snippet durch die Werte für Ihren Mandanten.

   import oci  
     
   from base64 import b64encode  
     
   ociMessageEndpoint = "<stream_message_endpoint>"  
   ociStreamOcid = "<stream_OCID>"  
   ociConfigFilePath = "<config_file_path>"  
   ociProfileName = "<config_file_profile_name>"  
     
   def produce_messages(client, stream_id):
     # Build up a PutMessagesDetails and publish some messages to the stream
     message_list = []
     for i in range(100):
         key = "messageKey" + str(i)
         value = "messageValue " + str(i)
         encoded_key = b64encode(key.encode()).decode()
         encoded_value = b64encode(value.encode()).decode()
         message_list.append(oci.streaming.models.PutMessagesDetailsEntry(key=encoded_key, value=encoded_value))  
     
     print("Publishing {} messages to the stream {} ".format(len(message_list), stream_id))
     messages = oci.streaming.models.PutMessagesDetails(messages=message_list)
     put_message_result = client.put_messages(stream_id, messages)
     
     # The put_message_result can contain some useful metadata for handling failures
     for entry in put_message_result.data.entries:
         if entry.error:
             print("Error ({}) : {}".format(entry.error, entry.error_message))
         else:
             print("Published message to partition {} , offset {}".format(entry.partition, entry.offset))
   
   config = oci.config.from_file(ociConfigFilePath, ociProfileName)
   stream_client = oci.streaming.StreamClient(config, service_endpoint=ociMessageEndpoint)
   
   # Publish some messages to the stream
   produce_messages(stream_client, ociStreamOcid)

3. Führen Sie im Verzeichnis wd den folgenden Befehl aus:

   python Producer.py

4. Zeigen Sie die neuesten Nachrichten an, die an den Stream gesendet wurden, um die neuesten Nachrichten anzuzeigen, die an den Stream gesendet wurden, um zu prüfen, ob die Produktion erfolgreich war.

1. Stellen Sie zunächst sicher, dass der Stream, aus dem Sie Nachrichten konsumieren möchten, Nachrichten enthält. Sie können eine Testnachricht mit der Konsole erstellen oder den Stream und die Nachrichten verwenden, die wir in diesem Schnellstart erstellt haben.
2. Öffnen Sie Ihren bevorzugten Editor, wie Visual Studio Code, im Verzeichnis wd. oci-sdk-Packages für Python sollten bereits in der aktuellen Python-Umgebung installiert sein, nachdem Sie die Schritte unter Voraussetzungen ausgeführt haben.

3. Erstellen Sie eine Datei namens Consumer.py im Verzeichnis wd mit dem folgenden Code. Ersetzen Sie die Werte der Variablen ociConfigFilePath, ociProfileName,ociStreamOcid und ociMessageEndpoint im folgenden Code-Snippet durch die Werte für Ihren Mandanten.

   import oci
   import time
   
   from base64 import b64decode
   
   ociMessageEndpoint = "<stream_message_endpoint>"  
   ociStreamOcid = "<stream_OCID>"  
   ociConfigFilePath = "<config_file_path>"  
   ociProfileName = "<config_file_profile_name>"  
   
   
   def get_cursor_by_group(sc, sid, group_name, instance_name):
       print(" Creating a cursor for group {}, instance {}".format(group_name, instance_name))
       cursor_details = oci.streaming.models.CreateGroupCursorDetails(group_name=group_name, instance_name=instance_name,
                                                                      type=oci.streaming.models.
                                                                      CreateGroupCursorDetails.TYPE_TRIM_HORIZON,
                                                                      commit_on_get=True)
       response = sc.create_group_cursor(sid, cursor_details)
       return response.data.value
   
   def simple_message_loop(client, stream_id, initial_cursor):
       cursor = initial_cursor
       while True:
           get_response = client.get_messages(stream_id, cursor, limit=10)
           # No messages to process. return.
           if not get_response.data:
               return
   
           # Process the messages
           print(" Read {} messages".format(len(get_response.data)))
           for message in get_response.data:
               if message.key is None:
                   key = "Null"
               else:
                   key = b64decode(message.key.encode()).decode()
               print("{}: {}".format(key,
                                     b64decode(message.value.encode()).decode()))
   
           # get_messages is a throttled method; clients should retrieve sufficiently large message
           # batches, as to avoid too many http requests.
           time.sleep(1)
           # use the next-cursor for iteration
           cursor = get_response.headers["opc-next-cursor"]
   
   
   config = oci.config.from_file(ociConfigFilePath, ociProfileName)
   stream_client = oci.streaming.StreamClient(config, service_endpoint=ociMessageEndpoint)
   
   # A cursor can be created as part of a consumer group.
   # Committed offsets are managed for the group, and partitions
   # are dynamically balanced amongst consumers in the group.
   group_cursor = get_cursor_by_group(stream_client, ociStreamOcid, "example-group", "example-instance-1")
   simple_message_loop(stream_client, ociStreamOcid, group_cursor)

4. Führen Sie im Verzeichnis wd den folgenden Befehl aus:

   python Consumer.py

5. Meldungen wie die Folgenden sollten angezeigt werden:

   Starting a simple message loop with a group cursor
   Creating a cursor for group example-group, instance example-instance-1
    Read 2 messages
   Null: Example Test Message 0
   Null: Example Test Message 0
    Read 2 messages
   Null: Example Test Message 0
   Null: Example Test Message 0
    Read 1 messages
   Null: Example Test Message 0
    Read 10 messages
   key 0: value 0
   key 1: value 1

   Hinweis
   
   Wenn Sie die Konsole zum Erzeugen einer Testnachricht verwendet haben, lautet der Schlüssel für jede Nachricht Null

Weitere Informationen finden Sie in den folgenden Ressourcen:

1. OCI-SDK für Python auf GitHub
2. Beispiele für OCI-SDK für Python für Streaming