Upload current snapshot.

Author: michaelsafyan

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/api/blob.py

@@ -1,10 +1,10 @@
+import base64
 from typing import Dict, Optional
 
-import base64
 
 class Blob(object):
     """Represents an opaque binary object and associated metadata.
-    
+
     This object conteptually has the following properties:
 
       - raw_bytes: the actual data (payload) of the Blob
@@ -13,9 +13,14 @@ class Blob(object):
          the object such as {"trace_id": "...", "span_id": "...", "filename": ...}
     """
 
-    def __init__(self, raw_bytes: bytes, content_type: Optional[str]=None, labels: Optional[Dict[str, str]]=None):
+    def __init__(
+        self,
+        raw_bytes: bytes,
+        content_type: Optional[str] = None,
+        labels: Optional[Dict[str, str]] = None,
+    ):
         """Initialize the blob with an explicit set of properties.
-        
+
         Args:
           raw_bytes: the required payload
           content_type: the MIME type describing the type of data in the payload
@@ -26,31 +31,41 @@ def __init__(self, raw_bytes: bytes, content_type: Optional[str]=None, labels: O
         self._labels = labels or {}
 
     @staticmethod
-    def from_data_uri(cls, uri: str, labels: Optional[dict]=None) -> 'Blob':
+    def from_data_uri(cls, uri: str, labels: Optional[dict] = None) -> "Blob":
         """Instantiate a blob from a 'data:...' URI.
 
-        Args:        
+        Args:
             uri: A URI in the 'data:' format. Supports a subset of 'data:' URIs
               that encode the data with the 'base64' extension and that include
               a content type. Should work with any normal 'image/jpeg', 'image/png',
               'application/pdf', 'audio/aac', and many others. DOES NOT SUPPORT
               encoding data as percent-encoded text (no "base64").
-    
+
             labels: Additional key/value data to include in the constructed Blob.
         """
-        if not uri.startswith('data:'):
-            raise ValueError('Invalid "uri"; expected "data:" prefix. Found: "{}"'.format(uri))
-        if not ';base64,' in uri:
-            raise ValueError('Invalid "uri"; expected ";base64," section. Found: "{}"'.format(uri))
-        data_prefix_len = len('data:')
+        if not uri.startswith("data:"):
+            raise ValueError(
+                'Invalid "uri"; expected "data:" prefix. Found: "{}"'.format(
+                    uri
+                )
+            )
+        if ";base64," not in uri:
+            raise ValueError(
+                'Invalid "uri"; expected ";base64," section. Found: "{}"'.format(
+                    uri
+                )
+            )
+        data_prefix_len = len("data:")
         after_data_prefix = uri[data_prefix_len:]
-        if ';' not in after_data_prefix:
-            raise ValueError('Invalid "uri"; expected ";" in URI. Found: "{}"'.format(uri))
-        content_type, remaining = after_data_prefix.split(';', 1)
-        while not remaining.startswith('base64,'):
-            _, remaining = remaining.split(';', 1)
-        assert remaining.startswith('base64,')
-        base64_len = len('base64,')
+        if ";" not in after_data_prefix:
+            raise ValueError(
+                'Invalid "uri"; expected ";" in URI. Found: "{}"'.format(uri)
+            )
+        content_type, remaining = after_data_prefix.split(";", 1)
+        while not remaining.startswith("base64,"):
+            _, remaining = remaining.split(";", 1)
+        assert remaining.startswith("base64,")
+        base64_len = len("base64,")
         base64_encoded_content = remaining[base64_len:]
         try:
             raw_bytes = base64.standard_b64decode(base64_encoded_content)

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/api/blob_uploader.py

@@ -2,13 +2,15 @@
 
 import abc
 
-from opentelemetry.instrumentation._blobupload.api.constants import NOT_UPLOADED
 from opentelemetry.instrumentation._blobupload.api.blob import Blob
+from opentelemetry.instrumentation._blobupload.api.constants import (
+    NOT_UPLOADED,
+)
 
 
 class BlobUploader(abc.ABC):
     """Pure abstract base class representing a component that does blob uploading."""
-    
+
     @abc.abstractmethod
     def upload_async(self, blob: Blob) -> str:
         return NOT_UPLOADED

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/api/content_type.py

@@ -1,19 +1,21 @@
 """Provides utilities for automatic content-type detection."""
 
+
 # Helper used to handle the possibility of optional 'magic' dependency
 # being unavailable for guessing the MIME type of raw bytes.
 class _FallBackModule(object):
     """Class that is shaped like the portion of 'magic' we need."""
 
     def from_buffer(self, raw_bytes, mime=True):
         """Fallback, subpar implementation of 'from_buffer'."""
-        return 'application/octet-stream'
+        return "application/octet-stream"
 
 
 # Set up '_module' to either use 'magic' or the fallback.
 _module = _FallBackModule()
 try:
     import magic
+
     _module = magic
 except ImportError:
     pass

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/api/labels.py

@@ -1,29 +1,34 @@
 """Provides utilities for providing basic identifying labels for blobs."""
 
+
 def generate_labels_for_span(trace_id: str, span_id: str) -> dict:
     """Returns metadata for a span."""
-    return {
-        'otel_type': 'span',
-        'trace_id': trace_id,
-        'span_id': span_id
-    }
+    return {"otel_type": "span", "trace_id": trace_id, "span_id": span_id}
 
 
-def generate_labels_for_event(trace_id: str, span_id: str, event_name: str) -> dict:
+def generate_labels_for_event(
+    trace_id: str, span_id: str, event_name: str
+) -> dict:
     """Returns metadata for an event."""
     result = generate_labels_for_span(trace_id, span_id)
-    result.update({
-        'otel_type': 'event',
-        'event_name': event_name,
-    })
+    result.update(
+        {
+            "otel_type": "event",
+            "event_name": event_name,
+        }
+    )
     return result
 
 
-def generate_labels_for_span_event(trace_id: str, span_id: str, event_name: str, event_index: int) -> dict:
+def generate_labels_for_span_event(
+    trace_id: str, span_id: str, event_name: str, event_index: int
+) -> dict:
     """Returns metadata for a span event."""
     result = generate_labels_for_event(trace_id, span_id, event_name)
-    result.update({
-        'otel_type': 'span_event',
-        'event_index': event_index,
-    })
+    result.update(
+        {
+            "otel_type": "span_event",
+            "event_index": event_index,
+        }
+    )
     return result

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/api/provider.py

@@ -1,11 +1,11 @@
-from typing import Optional
-
 import abc
 import logging
+from typing import Optional
 
 from opentelemetry.instrumentation._blobupload.api.blob import Blob
-from opentelemetry.instrumentation._blobupload.api.blob_uploader import BlobUploader
-
+from opentelemetry.instrumentation._blobupload.api.blob_uploader import (
+    BlobUploader,
+)
 
 _logger = logging.getLogger(__name__)
 
@@ -22,7 +22,7 @@ class BlobUploaderProvider(abc.ABC):
 
     def get_blob_uploader(self, use_case: Optional[str]) -> BlobUploader:
         """Returns a BlobUploader for the specified use case.
-        
+
         Args:
           use_case: An optional use case that describes what the uploader is for. This could
             name a particular package, class, or instrumentation. It is intended to allow
@@ -38,10 +38,14 @@ class _DefaultBlobUploaderProvider(BlobUploaderProvider):
     """Default provider used when none has been configured."""
 
     def get_blob_uploader(self, use_case: Optional[str]) -> BlobUploader:
-        use_case_formatted = '(None)'
+        use_case_formatted = "(None)"
         if use_case:
             use_case_formatted = use_case
-        _logger.warning('No BlobUploaderProvider configured; returning a no-op for use case {}'.format(use_case_formatted))
+        _logger.warning(
+            "No BlobUploaderProvider configured; returning a no-op for use case {}".format(
+                use_case_formatted
+            )
+        )
         return _NoOpBlobUploader()
 
 

opentelemetry-instrumentation/tests/_blobupload/api/test_content_type.py

@@ -1,19 +1,21 @@
 #! /usr/bin/env python3
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     import sys
-    sys.path.append('../../../src')
 
-from opentelemetry.instrumentation._blobupload.api import detect_content_type
-from PIL import Image
+    sys.path.append("../../../src")
 
 import io
 import unittest
 
+from PIL import Image
+
+from opentelemetry.instrumentation._blobupload.api import detect_content_type
+
 
 def create_test_image(format):
     """Helper for creating a PIL Image for verifying image format support."""
-    test_img = Image.new('RGB', (2, 2))
+    test_img = Image.new("RGB", (2, 2))
     output_buffer = io.BytesIO()
     test_img.save(output_buffer, format)
     result = output_buffer.getvalue()
@@ -23,31 +25,30 @@ def create_test_image(format):
 
 
 class TestContentType(unittest.TestCase):
-
     def test_detects_plaintext(self):
-        input = 'this is just regular text'
+        input = "this is just regular text"
         output = detect_content_type(input.encode())
-        self.assertEqual(output, 'text/plain')
+        self.assertEqual(output, "text/plain")
 
     def test_detects_json(self):
-        input = '''{
+        input = """{
             "this": {
                 "contains": "json"
             }
-        }'''
+        }"""
         output = detect_content_type(input.encode())
-        self.assertEqual(output, 'application/json')
+        self.assertEqual(output, "application/json")
 
     def test_detects_jpeg(self):
-        input = create_test_image('jpeg')
+        input = create_test_image("jpeg")
         output = detect_content_type(input)
-        self.assertEqual(output, 'image/jpeg')
+        self.assertEqual(output, "image/jpeg")
 
     def test_detects_png(self):
-        input = create_test_image('png')
+        input = create_test_image("png")
         output = detect_content_type(input)
-        self.assertEqual(output, 'image/png')
+        self.assertEqual(output, "image/png")
 
 
-if __name__ == '__main__':
-    unittest.main()
+if __name__ == "__main__":
+    unittest.main()