Completely migrated to PIL, the crop functinality has been simplified, #112

Author: pylover

sqlalchemy_media/attachments.py

@@ -7,6 +7,7 @@
 from typing import Hashable, Tuple, List, Iterable
 
 from sqlalchemy.ext.mutable import MutableList, MutableDict
+from PIL import Image as PilImage
 
 from .constants import MB, KB
 from .descriptors import AttachableDescriptor
@@ -808,15 +809,14 @@ def generate_thumbnail(
             width, height, ratio
         )
 
-        CompatibleImage = get_image_factory()
         # opening the original file
         thumbnail_buffer = io.BytesIO()
         store = self.get_store()
+        format_ = 'jpeg'
         with store.open(self.path) as original_file:
 
             # generating thumbnail and storing in buffer
-            img = CompatibleImage(file=original_file)
-            img.format = 'jpg'
+            img = PilImage.open(original_file)
 
             with img:
                 original_size = img.size
@@ -829,8 +829,8 @@ def generate_thumbnail(
                 width = int(width)
                 height = int(height)
 
-                img.resize(width, height)
-                img.save(file=thumbnail_buffer)
+                thumbnail_image = img.resize((width, height))
+                thumbnail_image.save(thumbnail_buffer, format_)
 
         thumbnail_buffer.seek(0)
         if self.thumbnails is None:

sqlalchemy_media/descriptors.py

@@ -6,7 +6,7 @@
 
 from .constants import KB
 from .exceptions import MaximumLengthIsReachedError, \
-    MinimumLengthIsNotReachedError, DescriptorOperationError
+    DescriptorOperationError
 from .helpers import is_uri, copy_stream
 from .mimetypes_ import guess_extension, guess_type
 from .typing_ import FileLike, Attachable
@@ -233,21 +233,9 @@ def get_header_buffer(self) -> bytes:
 
     def close(self, check_length=True) -> None:
         """
-        Closes the underlying file-object. and check for ``min_length``.
-
-        :param check_length: Check the minimum length of the stream and
-                             :class:`MinimumLengthIsNotReachedError` may be
-                             raised during close. default is `True`.
-
-        .. versionadded:: 0.8
-
-            - `check_length`
-
+        Do nothing here.
         """
-        if check_length:
-            pos = self.tell()
-            if self.min_length and self.min_length > pos:
-                raise MinimumLengthIsNotReachedError(self.min_length, pos)
+        pass
 
     def seekable(self) -> bool:
         """

sqlalchemy_media/exceptions.py

@@ -16,18 +16,6 @@ def __init__(self, max_length: int):
         )
 
 
-class MinimumLengthIsNotReachedError(SqlAlchemyMediaException):
-    """
-    Indicates the minimum allowed length is not reached.
-    """
-
-    def __init__(self, min_length, length=None):
-        super().__init__(
-            f'Cannot store files smaller than: {min_length: d} bytes,'
-            f'but the file length is: {length}'
-        )
-
-
 class ContextError(SqlAlchemyMediaException):
     """
     Exception related to :class:`.StoreManager`.

sqlalchemy_media/processors.py

@@ -7,7 +7,7 @@
 from .exceptions import ContentTypeValidationError, DimensionValidationError, \
     AspectRatioValidationError, AnalyzeError
 from .helpers import validate_width_height_ratio, deprecated
-from .mimetypes_ import guess_extension
+from .mimetypes_ import guess_extension, guess_type
 from .optionals import magic_mime_from_buffer, ensure_wand
 from .typing_ import Dimension
 
@@ -427,72 +427,56 @@ def __init__(self, fmt: str = None, width: int = None, height: int = None, crop=
         #   k: v if isinstance(v, str) else str(v) for k, v in crop.items()
         # }
 
+    def _update_context(self, img: PilImage, format_, context: dict):
+        mimetype = guess_type(f'a.{format_}'.lower())
+        context.update(
+            content_type=mimetype,
+            width=img.width,
+            height=img.height,
+            extension=guess_extension(mimetype)
+        )
+
     def process(self, descriptor: StreamDescriptor, context: dict):
 
-        # FIXME: rewrite this method
-        import pudb; pudb.set_trace()  # XXX BREAKPOINT
         # Copy the original info
         # generating thumbnail and storing in buffer
         img = PilImage.open(descriptor)
 
-        is_invalid_format = self.format is None or img.format == self.format
-        is_invalid_size = (
+        format_unchanged = self.format is None or img.format == self.format
+        size_unchanged = (
             (self.width is None or img.width == self.width) and
             (self.height is None or img.height == self.height)
         )
 
-        if self.crop is None and is_invalid_format and is_invalid_size:
-            img.close()
+        if self.crop is None and format_unchanged and size_unchanged:
+            self._update_context(img, img.format, context)
             descriptor.prepare_to_read(backend='memory')
             return
 
+        # Preserving format
+        format_ = self.format or img.format
+
         if 'length' in context:
             del context['length']
 
         # opening the original file
         output_buffer = io.BytesIO()
 
         # Changing dimension if required.
-        if self.width or self.height:
+        if not size_unchanged:
             width, height, _ = \
                 validate_width_height_ratio(self.width, self.height, None)
-            img.resize((
+            img = img.resize((
                 width(img.size) if callable(width) else width,
                 height(img.size) if callable(height) else height
             ))
 
         # Cropping
         if self.crop:
-            def get_key_for_crop_item(key, value):
-                crop_width_keys = ('width', 'left', 'right')
-                crop_keys = (
-                    'left', 'top', 'right', 'bottom', 'width', 'height'
-                )
-
-                if key in crop_keys and isinstance(value, str) \
-                        and '%' in value:
-                    return int(int(value[:-1]) / 100 * (
-                        img.width if key in crop_width_keys else img.height
-                    ))
-
-                return value
-
-            img.crop(**{
-                key: get_key_for_crop_item(key, value)
-                for key, value in self.crop.items()
-            })
-
-        img.save(output_buffer, format=self.format)
-
-        mimetype = img.get_format_mimetype()
-
-        context.update(
-            content_type=mimetype,
-            width=img.width,
-            height=img.height,
-            extension=guess_extension(mimetype)
-        )
+            img = img.crop(self.crop)
 
+        img.save(output_buffer, format=format_)
+        self._update_context(img, format_, context)
         output_buffer.seek(0)
         descriptor.replace(output_buffer, position=0, **context)
 

sqlalchemy_media/tests/test_file.py

@@ -7,7 +7,7 @@
 from sqlalchemy_media import File, StoreManager, ContentTypeValidator, \
     MagicAnalyzer
 from sqlalchemy_media.exceptions import MaximumLengthIsReachedError, \
-    MinimumLengthIsNotReachedError, ContentTypeValidationError
+    ContentTypeValidationError
 from sqlalchemy_media.tests.helpers import Json, TempStoreTestCase
 
 
@@ -162,12 +162,6 @@ class Person(self.Base):
         person1.cv = LimitedFile()
 
         with StoreManager(session):
-
-            # MaximumLengthIsReachedError, MinimumLengthIsNotReachedError
-            self.assertRaises(
-                MinimumLengthIsNotReachedError,
-                person1.cv.attach, BytesIO(b'less than 20 chars!')
-            )
             self.assertRaises(
                 MaximumLengthIsReachedError,
                 person1.cv.attach,

sqlalchemy_media/tests/test_image_processors.py

@@ -37,25 +37,57 @@ def test_resize_reformat(self):
             # Checking when not modifying stream.
             ctx = dict()
             ImageProcessor().process(d, ctx)
-            self.assertFalse(len(ctx))
+            self.assertEqual(
+                {
+                    'content_type': 'image/jpeg',
+                    'width': 640,
+                    'height': 480,
+                    'extension': '.jpg'
+                }.items(),
+                ctx.items()
+            )
 
             # Checking when not modifying stream.
             ImageProcessor(fmt='jpeg').process(d, ctx)
-            self.assertFalse(len(ctx))
+            self.assertEqual(
+                {
+                    'content_type': 'image/jpeg',
+                    'width': 640,
+                    'height': 480,
+                    'extension': '.jpg'
+                }.items(),
+                ctx.items()
+            )
 
             ImageProcessor(fmt='jpeg', width=640).process(d, ctx)
-            self.assertFalse(len(ctx))
+            self.assertEqual(
+                {
+                    'content_type': 'image/jpeg',
+                    'width': 640,
+                    'height': 480,
+                    'extension': '.jpg'
+                }.items(),
+                ctx.items()
+            )
+
 
             ImageProcessor(fmt='jpeg', height=480).process(d, ctx)
-            self.assertFalse(len(ctx))
+            self.assertEqual(
+                {
+                    'content_type': 'image/jpeg',
+                    'width': 640,
+                    'height': 480,
+                    'extension': '.jpg'
+                }.items(),
+                ctx.items()
+            )
 
-"""
     def test_crop(self):
         with AttachableDescriptor(self.cat_jpeg) as d:
             # Checking when not modifying stream.
             ctx = dict()
             ImageProcessor(
-                crop=dict(width='50%', height='50%', gravity='center')
+                crop=(160, 120, 480, 360)
             ).process(d, ctx)
             ctx = dict()
             ImageAnalyzer().process(d, ctx)
@@ -68,22 +100,6 @@ def test_crop(self):
                 }
             )
 
-        # With integer values
-        with AttachableDescriptor(self.cat_jpeg) as d:
-            # Checking when not modifying stream.
-            ctx = dict()
-            ImageProcessor(crop=dict(width=100)).process(d, ctx)
-            ctx = dict()
-            ImageAnalyzer().process(d, ctx)
-            self.assertDictEqual(
-                ctx,
-                {
-                    'content_type': 'image/jpeg',
-                    'width': 100,
-                    'height': 480,
-                }
-            )
 
-"""
 if __name__ == '__main__':  # pragma: no cover
     unittest.main()