У меня есть пользовательское поле model_field в проекте. Я загружаю файл excel с данными, в котором храню ссылки на фотографии(такого типа
https://url?token).Я выполняю обработку Excel с помощью pandas и сохраняю данные в базе данных через сериализатор. Как я могу открыть ссылку и получить изображение, чтобы записать его в это поле модели.
Это пользовательское поле в model_field.py
class PsImageModelFieldBase(JSONField):
def __init__(self, ps_image_class, **kwargs):
assert issubclass(ps_image_class, BasePsImage)
self.ps_image_class = ps_image_class
super().__init__(**kwargs)
def deconstruct(self):
name, path, args, kwargs = super().deconstruct()
args.append(self.ps_image_class)
return name, path, args, kwargs
def create_object(self, kwargs):
if kwargs is None:
return None
try:
if isinstance(kwargs, BasePsImage):
return kwargs
return self.ps_image_class(**kwargs)
except TypeError:
return kwargs
class PsImageModelField(PsImageModelFieldBase):
def get_prep_value(self, value):
if value is not None and isinstance(value, (dict, list, BasePsImage)):
value = self.create_object(value).as_json
return super().get_prep_value(value)
def from_db_value(self, value, expression, connection):
return self.create_object(value)
```
image.py
```
import os
import re
from django.conf import settings
from common.utils import camel_to_snake
from photoservice.service import PhotoService
class InvalidFormatError(Exception):
pass
_default_photo_service = PhotoService()
def _validate_format(format_, format_regexes):
is_valid = any(regex.match(format_) for regex in format_regexes)
return is_valid
def _validate_versions(class_name, versions, format_regexes):
for fname, format_ in versions.items():
if not _validate_format(format_, format_regexes):
raise InvalidFormatError(f'{class_name}: version="{fname}": "{format_}" is invalid')
class PsImageMeta(type):
"""
Metaclass for default formats validation.
"""
def __new__(mcs, name, bases, attrs):
format_regexes = attrs.get('_format_regexes') or next(b._format_regexes for b in bases)
_validate_versions(name, attrs['default_versions'], format_regexes)
return type.__new__(mcs, name, bases, attrs)
class BasePsImage(metaclass=PsImageMeta):
_default_photo_service = _default_photo_service
_format_regexes = (
re.compile(r'^fit_s[^0]\d{,5}x[^0]\d{,5}q\d{1,2}.jpg$'),
re.compile(r'^(?:min|max)_s[^0]\d{,5}q\d{1,2}.jpg$'),
re.compile(r'^same_q\d{1,2}.jpg$'),
)
default_versions = {}
include_original_link = False
new_key_template = None
key_prefix = None
_min_width = 100
_min_height = 100
def __init__(self, key, versions=None, webhook=None, redirect=None, replace=None,
extra_data=None, service=None, **kwargs):
assert isinstance(key, str)
assert versions is None or isinstance(versions, dict)
assert self.new_key_template is not None, 'new key template must be set'
if versions:
self._validate_versions(versions)
self._key = key
self._versions = versions or self.default_versions
self._service = service or self._default_photo_service
self._webhook = webhook
self._redirect = redirect
self._replace = replace
self._extra_data = extra_data
def __str__(self):
return f'{self.__class__.__name__}(key={self._key!r})'
def __repr__(self):
return self.__str__()
@classmethod
def new_key(cls, obj):
return os.path.join(
cls.get_key_prefix(),
cls.new_key_template.format(
class_name=obj.__class__.__name__,
obj_id=obj.id,
)
)
@classmethod
def get_key_prefix(cls):
return cls.key_prefix or camel_to_snake(cls.__name___)
@classmethod
def get_extra_data(cls, obj):
return {
'obj_id': obj.id,
'obj_type': obj.__class__.__name__
}
@classmethod
def for_object(cls, obj, **kwargs):
return cls(key=cls.new_key(obj), extra_data=cls.get_extra_data(obj), **kwargs)
@classmethod
def from_token(cls, token, service=None):
service = service or cls._default_photo_service
data = service.decode_token(token, raise_=True)
return cls(**data)
@property
def key(self):
return self._key
@key.setter
def key(self, value):
self._key = value
@property
def versions(self):
return self._versions
@property
def extra_data(self):
return self._extra_data
@property
def links(self):
links = {k: self.link_for(ver) for k, ver in self._versions.items()}
if self.include_original_link:
links['original'] = self.link_for('original' + settings.PHOTOSERVICE_ORIGINAL_IMG_EXT)
return links
@property
def upload_link(self):
return self._service.upload_link(
self._key, self._versions, webhook=self._webhook, redirect=self._redirect,
replace=self._replace, extra_data=self._extra_data, min_width=self._min_width, min_height=self._min_height
)
@property
def as_json(self):
"""
:return: JSON serializable dict which will be stored in DB
"""
return {
'key': self._key,
'versions': self._versions
}
def link_for(self, version='original'):
assert version in self._versions.values() or version in ('original', 'original' + settings.PHOTOSERVICE_ORIGINAL_IMG_EXT)
return self._service.download_link(self._key, version)
def _validate_format(self, format_):
return _validate_format(format_, self._format_regexes)
def _validate_versions(self, versions):
_validate_versions(self.__class__.__name__, versions, self._format_regexes)
models.py
class BasePhoto(models.Model):
photo = PsImageModelField(ps_image_class=image.SurveyImage)
order = models.IntegerField(default=0)
@property
def links(self):
return self.photo.links
class Meta:
abstract = True
ordering = ('order', 'id')
Простой
Средний