Python для облака

Почему Python

CRC использует Python для серверной части по нескольким причинам:

1. Самый популярный язык для облака -- все три провайдера имеют первоклассные Python SDK
2. Простой синтаксис -- легко читать и писать даже новичкам
3. Serverless-дружественный -- быстрый cold start, маленький размер пакетов
4. Богатая экосистема -- boto3, azure-sdk, google-cloud-* для всех сервисов

Необходимый минимум Python

Функции

# Обычная функция
def get_visitor_count(table_name: str) -> int:
    """Retrieve current visitor count from database."""
    # Implementation here
    return count

# Lambda-функция (не путать с AWS Lambda!)
increment = lambda x: x + 1

# Функция с default параметром
def format_response(body: dict, status_code: int = 200) -> dict:
    return {
        'statusCode': status_code,
        'body': json.dumps(body)
    }

Работа с JSON

import json

# Python dict -> JSON string
data = {'count': 42, 'message': 'Hello'}
json_string = json.dumps(data)
# '{"count": 42, "message": "Hello"}'

# JSON string -> Python dict
parsed = json.loads(json_string)
print(parsed['count'])  # 42

# JSON с отступами (для отладки)
pretty = json.dumps(data, indent=2, ensure_ascii=False)

Обработка ошибок

import json
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def handler(event, context):
    try:
        count = get_and_increment_count()
        logger.info(f"Visitor count updated to {count}")

        return {
            'statusCode': 200,
            'body': json.dumps({'count': count})
        }
    except KeyError as e:
        logger.error(f"Missing key: {e}")
        return {
            'statusCode': 400,
            'body': json.dumps({'error': f'Missing key: {e}'})
        }
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': 'Internal server error'})
        }

Словари и условия

# Работа со словарями (основа для DynamoDB)
item = {
    'id': 'visitor-counter',
    'count': 0
}

# Безопасное получение значения
count = item.get('count', 0)  # 0 если ключа нет

# Условное обновление
if count > 1000:
    notification = 'Milestone reached!'

Type hints

from typing import Any

def lambda_handler(event: dict[str, Any], context: Any) -> dict[str, Any]:
    """AWS Lambda handler with type hints."""
    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*'
        },
        'body': json.dumps({'count': 42})
    }

AWS SDK: boto3

Установка и настройка

pip install boto3

import boto3

# Клиент создаётся автоматически с credentials из:
# 1. Переменные окружения (в Lambda автоматически)
# 2. ~/.aws/credentials (локально)
# 3. IAM Role (рекомендуется для Lambda)

# Resource API (высокоуровневый)
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('visitor-counter')

# Client API (низкоуровневый)
client = boto3.client('dynamodb')

Работа с DynamoDB через boto3

import boto3
import json
from decimal import Decimal

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('visitor-counter')

def lambda_handler(event, context):
    # Атомарное увеличение счётчика
    response = table.update_item(
        Key={'id': 'visitors'},
        UpdateExpression='ADD visitor_count :inc',
        ExpressionAttributeValues={':inc': Decimal('1')},
        ReturnValues='UPDATED_NEW'
    )

    count = int(response['Attributes']['visitor_count'])

    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*'
        },
        'body': json.dumps({'count': count})
    }

Важные особенности boto3

• Decimal: DynamoDB использует Decimal вместо float -- json.dumps не умеет сериализовать Decimal, нужно конвертировать в int
• Regions: По умолчанию boto3 использует регион из переменной окружения AWS_DEFAULT_REGION
• Retries: boto3 автоматически повторяет запросы при throttling

Azure SDK

Установка

pip install azure-functions azure-cosmos

Azure Functions handler

import azure.functions as func
import json
import logging
from azure.cosmos import CosmosClient

# Connection string from environment variable
COSMOS_CONNECTION = os.environ['COSMOS_CONNECTION_STRING']
client = CosmosClient.from_connection_string(COSMOS_CONNECTION)
database = client.get_database_client('crc-database')
container = database.get_container_client('visitors')

def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Visitor counter function triggered')

    try:
        # Read current count
        item = container.read_item(item='counter', partition_key='counter')
        item['count'] += 1

        # Update count
        container.upsert_item(item)

        return func.HttpResponse(
            json.dumps({'count': item['count']}),
            mimetype='application/json',
            headers={'Access-Control-Allow-Origin': '*'}
        )
    except Exception as e:
        logging.error(f'Error: {e}')
        return func.HttpResponse(
            json.dumps({'error': 'Internal error'}),
            status_code=500,
            mimetype='application/json'
        )

Google Cloud SDK

Установка

pip install google-cloud-firestore functions-framework

Cloud Functions handler

import functions_framework
from google.cloud import firestore
import json

db = firestore.Client()

@functions_framework.http
def visitor_counter(request):
    """HTTP Cloud Function for visitor counter."""
    # Handle CORS preflight
    if request.method == 'OPTIONS':
        headers = {
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Methods': 'GET',
            'Access-Control-Max-Age': '3600'
        }
        return ('', 204, headers)

    headers = {
        'Access-Control-Allow-Origin': '*',
        'Content-Type': 'application/json'
    }

    try:
        # Atomic increment using transaction
        counter_ref = db.collection('counters').document('visitors')

        @firestore.transactional
        def increment(transaction):
            snapshot = counter_ref.get(transaction=transaction)
            if snapshot.exists:
                new_count = snapshot.get('count') + 1
            else:
                new_count = 1
            transaction.set(counter_ref, {'count': new_count})
            return new_count

        transaction = db.transaction()
        count = increment(transaction)

        return (json.dumps({'count': count}), 200, headers)

    except Exception as e:
        return (json.dumps({'error': str(e)}), 500, headers)

Сравнение SDK

Локальная разработка и тестирование

Запуск Lambda локально

# С помощью SAM CLI
pip install aws-sam-cli
sam local invoke VisitorCounterFunction -e events/test.json

# Тестовое событие
# events/test.json
{
  "httpMethod": "GET",
  "path": "/count"
}

Запуск Azure Functions локально

# Azure Functions Core Tools
pip install azure-functions-core-tools
func start

Запуск Cloud Functions локально

# functions-framework
pip install functions-framework
functions-framework --target=visitor_counter --port=8080