Install boto3 in a Python environment and record the installed version. Configure an AWS role or SSO profile with rekognition:DetectLabels permission. Set AWS_REGION and IMAGE_PATH; optionally select AWS_PROFILE. Save as rekognition_smoke.py and run python rekognition_smoke.py. No account, credentials or cloud call is provided by this page.
Keep credentials in your local or server-side credential provider. Running this sample sends an image to provider cloud and may consume account usage.
View runnable Python starter · rekognition_smoke.py
"""One authorized JPEG/PNG sent to AWS DetectLabels; no automatic retry."""
import json
import math
import os
import sys
from datetime import datetime, timezone
from importlib.metadata import version
from pathlib import Path
from time import monotonic
def normalize(raw):
if not isinstance(raw, dict) or not isinstance(raw.get('Labels'), list):
raise ValueError('unexpected_schema')
output = []
for item in raw['Labels']:
if not isinstance(item, dict) or not isinstance(item.get('Name'), str):
raise ValueError('unexpected_schema')
confidence = item.get('Confidence')
if type(confidence) not in (int, float) or not math.isfinite(confidence) or not 0 <= confidence <= 100:
raise ValueError('unexpected_schema')
output.append({'label': item['Name'], 'score': confidence / 100})
model = raw.get('LabelModelVersion')
if model is not None and not isinstance(model, str):
raise ValueError('unexpected_schema')
return output, model
def create_client(region):
import boto3
from botocore.config import Config
settings = Config(connect_timeout=5, read_timeout=30,
retries={'mode': 'standard', 'total_max_attempts': 1})
# Standard AWS credentials chain; no access keys in source code.
return boto3.client('rekognition', region_name=region, config=settings)
def classify_error(exc):
code = getattr(exc, 'response', {}).get('Error', {}).get('Code')
if code in ('AccessDeniedException', 'UnrecognizedClientException', 'InvalidSignatureException', 'ExpiredTokenException'):
return 'authorization_failure'
if code in ('ThrottlingException', 'ProvisionedThroughputExceededException'):
return 'throttled'
if code in ('InvalidImageFormatException', 'ImageTooLargeException', 'InvalidParameterException'):
return 'provider_input_failure'
if type(exc).__name__ in ('NoCredentialsError', 'PartialCredentialsError', 'ProfileNotFound'):
return 'credentials_missing'
if type(exc).__name__ in ('ReadTimeoutError', 'ConnectTimeoutError', 'EndpointConnectionError'):
return 'transport_failure'
return 'provider_or_transport_failure'
def run(client_factory=create_client):
start = monotonic()
result = {'schema_version': '1.0.0', 'provider': 'Amazon Rekognition', 'status': 'error',
'model_version': None, 'data': None, 'error': None,
'captured_at': datetime.now(timezone.utc).isoformat()}
try:
region = os.environ['AWS_REGION'].strip()
filename = Path(os.environ['IMAGE_PATH'])
if not region or filename.suffix.lower() not in ('.jpg', '.jpeg', '.png'):
raise ValueError('configuration')
# Bounded read: this starter's cap is 4 MiB, not a claimed service limit.
with filename.open('rb') as image:
content = image.read(4 * 1024 * 1024 + 1)
if not content or len(content) > 4 * 1024 * 1024:
raise ValueError('input_size')
client = client_factory(region)
try:
raw = client.detect_labels(Image={'Bytes': content}, MaxLabels=20,
MinConfidence=70, Features=['GENERAL_LABELS'])
result['data'], result['model_version'] = normalize(raw)
finally:
client.close()
result['region'] = region
result['status'] = 'ok'
except (KeyError, OSError):
result['error'] = 'configuration'
except (ValueError, TypeError):
result['error'] = 'configuration_or_schema'
except ImportError:
result['error'] = 'dependency_missing'
except Exception as exc:
result['error'] = classify_error(exc)
if result['status'] != 'ok':
result['data'] = None
result['elapsed_ms'] = round((monotonic() - start) * 1000)
return result
if __name__ == '__main__':
output = run()
print(json.dumps(output, ensure_ascii=False, allow_nan=False))
sys.exit(0 if output['status'] == 'ok' else 1)