OCI Developer Associate - Application Deployment and Integration¶

from oci.queue import QueueClient
from oci.queue.models import PutMessagesDetails, PutMessagesDetailsEntry

# Initialize client
queue_client = QueueClient(config)

# Send message
put_messages_response = queue_client.put_messages(
    queue_id="ocid1.queue...",
    put_messages_details=PutMessagesDetails(
        messages=[
            PutMessagesDetailsEntry(
                content="Task data here",
                metadata={
                    "priority": "high",
                    "task_type": "image_processing"
                }
            )
        ]
    )
)

print(f"Message ID: {put_messages_response.data.entries[0].id}")

import time

def process_queue_messages():
    while True:
        # Get messages (up to 20)
        get_messages_response = queue_client.get_messages(
            queue_id="ocid1.queue...",
            visibility_in_seconds=30,  # Visibility timeout
            timeout_in_seconds=10,     # Long polling
            limit=20
        )

        for message in get_messages_response.data.messages:
            try:
                # Process message
                process_task(message.content)

                # Delete message after successful processing
                queue_client.delete_message(
                    queue_id="ocid1.queue...",
                    message_receipt=message.receipt
                )

            except Exception as e:
                print(f"Error processing message: {e}")
                # Message will become visible again after timeout

        if not get_messages_response.data.messages:
            time.sleep(1)  # Wait before next poll

from oci.streaming import StreamClient
from oci.streaming.models import PutMessagesDetails, PutMessagesDetailsEntry
import json

stream_client = StreamClient(config, service_endpoint=stream_endpoint)

def send_event(event_data, partition_key):
    # Publish message
    put_messages_result = stream_client.put_messages(
        stream_id="ocid1.stream...",
        put_messages_details=PutMessagesDetails(
            messages=[
                PutMessagesDetailsEntry(
                    key=partition_key,  # Determines partition
                    value=json.dumps(event_data).encode('utf-8')
                )
            ]
        )
    )

    return put_messages_result.data.entries[0]

# Send events
send_event({"user_id": 123, "action": "login"}, partition_key="user-123")
send_event({"user_id": 456, "action": "purchase"}, partition_key="user-456")

from oci.streaming.models import CreateCursorDetails

def consume_stream():
    # Create cursor (starting point)
    cursor_details = CreateCursorDetails(
        partition="0",
        type="TRIM_HORIZON"  # Start from beginning
        # type="LATEST"      # Start from newest
        # type="AT_OFFSET"   # Start from specific offset
    )

    cursor = stream_client.create_cursor(
        stream_id="ocid1.stream...",
        create_cursor_details=cursor_details
    )

    current_cursor = cursor.data.value

    # Consume messages
    while True:
        get_messages = stream_client.get_messages(
            stream_id="ocid1.stream...",
            cursor=current_cursor,
            limit=100
        )

        # Process messages
        for message in get_messages.data:
            process_event(json.loads(message.value.decode('utf-8')))

        # Update cursor for next batch
        current_cursor = get_messages.headers.get("opc-next-cursor")

        if not get_messages.data:
            time.sleep(1)

# Multiple consumers in a group share partitions
from oci.streaming.models import CreateGroupCursorDetails

group_cursor = stream_client.create_group_cursor(
    stream_id="ocid1.stream...",
    create_group_cursor_details=CreateGroupCursorDetails(
        group_name="processor-group",
        instance_name="processor-1",
        type="TRIM_HORIZON",
        commit_on_get=True  # Auto-commit offsets
    )
)

# Python example with OpenTelemetry
from opentelemetry import trace
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger import JaegerExporter

# Setup tracing
trace.set_tracer_provider(TracerProvider())
jaeger_exporter = JaegerExporter(
    agent_host_name="jaeger",
    agent_port=6831,
)
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

# Instrument Flask app
FlaskInstrumentor().instrument_app(app)

# Manual span creation
tracer = trace.get_tracer(__name__)

@app.route('/api/data')
def get_data():
    with tracer.start_as_current_span("fetch-data"):
        # Span automatically includes timing
        data = database.query()
        return jsonify(data)

import cx_Oracle
import os

# Connection using wallet
def connect_to_adb():
    # Wallet location
    os.environ['TNS_ADMIN'] = '/path/to/wallet'

    # Connection string from tnsnames.ora
    dsn = cx_Oracle.makedsn(
        "adb.us-phoenix-1.oraclecloud.com",
        1522,
        service_name="dbname_high"
    )

    # Connect
    connection = cx_Oracle.connect(
        user="ADMIN",
        password=os.environ.get('DB_PASSWORD'),
        dsn=dsn
    )

    return connection

# Using connection pool (recommended)
def create_connection_pool():
    pool = cx_Oracle.SessionPool(
        user="ADMIN",
        password=os.environ.get('DB_PASSWORD'),
        dsn=dsn,
        min=2,
        max=10,
        increment=1,
        threaded=True
    )

    return pool

# Application usage
pool = create_connection_pool()

@app.route('/api/users')
def get_users():
    connection = pool.acquire()
    try:
        cursor = connection.cursor()
        cursor.execute("SELECT * FROM users")
        users = cursor.fetchall()
        return jsonify(users)
    finally:
        pool.release(connection)