Skip to content

Commit

Permalink
Auto coercion, now you can assign the file directly to the model's at…
Browse files Browse the repository at this point in the history
…tribute, closes #98
  • Loading branch information
pylover committed Mar 24, 2018
1 parent d588fe6 commit 609ab5e
Show file tree
Hide file tree
Showing 2 changed files with 84 additions and 41 deletions.
77 changes: 36 additions & 41 deletions sqlalchemy_media/attachments.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,15 +74,6 @@ def _listen_on_attribute(cls, attribute, coerce, parent_cls):
StoreManager.observe_attribute(attribute)
super()._listen_on_attribute(attribute, coerce, parent_cls)

@classmethod
def _assert_type(cls, value) -> None:
"""
Checking attachment type, raising :exc:`TypeError` if the value is not derived from :class:`.Attachment`.
"""
if not isinstance(value, cls):
raise TypeError('Value type must be subclass of %s, but it\'s: %s' % (cls, type(value)))

@classmethod
def coerce(cls, key, value) -> 'Attachment':
"""
Expand All @@ -91,15 +82,19 @@ def coerce(cls, key, value) -> 'Attachment':
.. seealso:: :meth:`sqlalchemy.ext.mutable.MutableDict.coerce`
"""
if value is not None and not isinstance(value, dict):
try:
cls._assert_type(value)
except TypeError:
if cls.__auto_coercion__:
return cls.create_from(value)
raise
if value is None or isinstance(value, (cls, dict)):
return super().coerce(key, value)

return super().coerce(key, value)
if cls.__auto_coercion__:
if not isinstance(value, (tuple, list)):
value = (value, )

return cls.create_from(*value)

raise TypeError(
'Value type must be subclass of %s or a tuple(file, mime, [filename]),'
'but it\'s: %s' % (cls, type(value))
)

@classmethod
def create_from(cls, *args, **kwargs):
Expand Down Expand Up @@ -168,7 +163,7 @@ def path(self) -> str:
def filename(self) -> str:
"""
The filename used to store the attachment in the storage with this format::
'{self.__prefix__}-{self.key}{self.suffix}{if self.extension else ''}'
:type: str
Expand Down Expand Up @@ -250,14 +245,14 @@ def reproducible(self) -> bool:
def copy(self) -> 'Attachment':
"""
Copy this object using deepcopy.
"""
return self.__class__(copy.deepcopy(self))

def get_store(self) -> Store:
"""
Returns the :class:`sqlalchemy_media.stores.Store` instance, which this file is stored on.
"""
store_manager = StoreManager.get_current_store_manager()
return store_manager.get(self.store_id)
Expand All @@ -269,7 +264,7 @@ def delete(self) -> None:
.. warning:: This operation can not be roll-backed.So if you want to delete a file, just set it to
:const:`None` or set it by new :class:`.Attachment` instance, while passed ``delete_orphan=True``
in :class:`.StoreManager`.
"""
self.get_store().delete(self.path)

Expand Down Expand Up @@ -319,7 +314,7 @@ def attach(self, attachable: Attachable, content_type: str = None, original_file
:param attachable: file-like object, filename or URL to attach.
:param content_type: If given, the content-detection is suppressed.
:param original_filename: Original name of the file, if available, to append to the end of the the filename,
:param original_filename: Original name of the file, if available, to append to the end of the the filename,
useful for SEO, and readability.
:param extension: The file's extension, is available.else, tries to guess it by content_type
:param store_id: The store id to store this file on. Stores must be registered with appropriate id via
Expand Down Expand Up @@ -427,13 +422,13 @@ def get_objects_to_delete(self):

def get_orphaned_objects(self):
"""
this method will be always called by the store when adding the ``self`` to the orphaned list. so subclasses
of the :class:`.Attachment` has a chance to add other related objects into the orphaned list and schedule it
for delete. for example the :class:`.Image` class can schedule it's thumbnails for deletion also.
this method will be always called by the store when adding the ``self`` to the orphaned list. so subclasses
of the :class:`.Attachment` has a chance to add other related objects into the orphaned list and schedule it
for delete. for example the :class:`.Image` class can schedule it's thumbnails for deletion also.
:return: An iterable of :class:`.Attachment` to mark as orphan.
.. versionadded:: 0.11.0
"""
return iter([])

Expand Down Expand Up @@ -477,18 +472,18 @@ class Person(BaseModel):
def observe_item(self, item):
"""
A simple monkeypatch to instruct the children to notify the parent if contents are changed:
From `sqlalchemy mutable documentation:
<http://docs.sqlalchemy.org/en/latest/orm/extensions/mutable.html#sqlalchemy.ext.mutable.MutableList>`_
Note that MutableList does not apply mutable tracking to the values themselves inside the list. Therefore
it is not a sufficient solution for the use case of tracking deep changes to a recursive mutable structure,
such as a JSON structure. To support this use case, build a subclass of MutableList that provides
appropriate coercion to the values placed in the dictionary so that they too are “mutable”, and emit events
Note that MutableList does not apply mutable tracking to the values themselves inside the list. Therefore
it is not a sufficient solution for the use case of tracking deep changes to a recursive mutable structure,
such as a JSON structure. To support this use case, build a subclass of MutableList that provides
appropriate coercion to the values placed in the dictionary so that they too are “mutable”, and emit events
up to their parent structure.
:param item: The item to observe
:return:
:return:
"""
item = self.__item_type__.coerce(None, item)
item._parents = self._parents
Expand Down Expand Up @@ -574,7 +569,7 @@ class Person(BaseModel):
me = Person()
me.files = MyDict()
me.files['original'] = MyAttachment.create_from(any_file)
"""

@classmethod
Expand Down Expand Up @@ -627,7 +622,7 @@ def __setitem__(self, key, value):
class File(Attachment):
"""
Representing an attached file. Normally if you want to store any file, this class is the best choice.
"""

__directory__ = 'files'
Expand Down Expand Up @@ -675,7 +670,7 @@ def attach(self, *args, dimension: Dimension=None, **kwargs):
:param kwargs: The same as the: :meth:`.Attachment.attach`.
:return: The same as the: :meth:`.Attachment.attach`.
"""
if dimension:
kwargs['width'], kwargs['height'] = dimension
Expand Down Expand Up @@ -854,9 +849,9 @@ def get_objects_to_delete(self):

def get_orphaned_objects(self):
"""
Mark thumbnails for deletion when the :class:`.Image` is being deleted.
Mark thumbnails for deletion when the :class:`.Image` is being deleted.
:return: An iterable of :class:`.Thumbnail` to mark as orphan.
.. versionadded:: 0.11.0
"""
Expand All @@ -870,7 +865,7 @@ def get_orphaned_objects(self):
class ImageList(AttachmentList):
"""
Used to create a collection of :class:`.Image`es
.. versionadded:: 0.11.0
"""

Expand Down
48 changes: 48 additions & 0 deletions sqlalchemy_media/tests/test_generic_assignment.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,48 @@
import unittest
import io
from os.path import join, exists

from sqlalchemy import Column, Integer

from sqlalchemy_media.attachments import File
from sqlalchemy_media.stores import StoreManager
from sqlalchemy_media.tests.helpers import Json, TempStoreTestCase


class AutoCoerceFile(File):
__auto_coercion__ = True


class GenericAssignmentTestCase(TempStoreTestCase):

def test_file_assignment(self):

class Person(self.Base):
__tablename__ = 'person'
id = Column(Integer, primary_key=True)
cv = Column(AutoCoerceFile.as_mutable(Json))

session = self.create_all_and_get_session()
person1 = Person()
resume = io.BytesIO(b'This is my resume')
with StoreManager(session):
person1.cv = resume
self.assertIsNone(person1.cv.content_type)
self.assertIsNone(person1.cv.extension)
self.assertTrue(exists(join(self.temp_path, person1.cv.path)))

person1.cv = resume, 'text/plain'
self.assertEqual(person1.cv.content_type, 'text/plain')
self.assertEqual(person1.cv.extension, '.txt')
self.assertTrue(exists(join(self.temp_path, person1.cv.path)))

person1.cv = resume, 'text/plain', 'myfile.note'
self.assertEqual(person1.cv.content_type, 'text/plain')
self.assertEqual(person1.cv.extension, '.note')
self.assertTrue(exists(join(self.temp_path, person1.cv.path)))



if __name__ == '__main__': # pragma: no cover
unittest.main()

0 comments on commit 609ab5e

Please sign in to comment.