Django REST 프레임워크의 다양한 팁 7.가져오기 내보내기
Django REST 프레임워크의 다양한 팁[디렉터리 인덱스]
도입 도출은 cms에서 매우 자주 사용하는 기능으로 통용되는 것을 생각하고 최종적으로django-import-export를 선택했다. 비록 이 물건은 처음에admin에 사용하려고 했지만 사용하기가 번거롭지만 통용되는 물건을 만들 수 있고 rest와 비슷한serializer보다 사용할 수 있다.
django-import-export==0.4.2 문서
볼 원본 cd 당신의virtualenv/local/lib/python2.7/site-packages/import_export
resources.py instance_loaders.py
먼저 용법을 보다
view를 통해 알 수 있듯이 코드는 이곳에서 매우 깨끗하고 정상적인 restframework의api와 별 차이가 없다.
class SchoolExportView(ExportMixin, GenericAPIView):
serializer_class = SchoolSerializer
permission_classes = (IsAuthenticated, ModulePermission)
queryset = School.objects.filter(is_active=True).order_by('-id')
resource_class = SchoolResource
filter_backends = (filters.DjangoFilterBackend, filters.SearchFilter)
filter_class = SchoolFilter
search_fields = ('name', 'contact')
module_perms = ['school.school']
class SchoolImportView(ImportMixin, GenericAPIView):
serializer_class = SchoolSerializer
permission_classes = (IsAuthenticated, ModulePermission)
queryset = School.objects.filter(is_active=True).order_by('-id')
resource_class = SchoolResource
module_perms = ['school.school']Mixin
class ExportMixin(object):
@GET('filename', type='string', default='download.xls')
@GET('format', type='string', default='xls', validators='in: xls,xlsx')
@GET('empty', type='bool', default=False)
def get(self, request, format, filename, empty):
queryset = None
if not empty:
queryset = self.filter_queryset(self.get_queryset())
resourse = self.resource_class()
export_data = resourse.export(queryset, empty)
return attachment_response(getattr(export_data, format), filename=filename)
class ImportMixin(object):
@POST('file', validators='required')
def post(self, request, file):
import_file = request.FILES['file']
resource = self.resource_class()
extra_data = {} if not hasattr(self, 'get_resoucre_extra_data') else self.get_resoucre_extra_data()
resource.set_extra_data(extra_data)
dataset = resource.get_dataset(import_file)
result = resource.import_data(dataset, use_transactions=True, raise_errors=True)
return Response()중점은 Resource를 실현하는 것이다. 먼저 export를 말해라.
export는 매우 간단하기 때문에 export를 먼저 말하고 데모를 먼저 본다. (export만 쓴다)
# -*- coding: utf-8 -*-
from __future__ import absolute_import
from import_export import resources
from school.models import School
class SchoolResource(resources.ModelResource):
def dehydrate_category(self, school):
if school.category == School.MIDDLE_SCHOOL:
return u' '
elif school.category == School.COLLEGE:
return u' '
return ''
def get_export_headers(self):
return [u' ', u' ', u' ', u' ', u' ', u' ',
u' ', u' ', u' ']
class Meta:
model = School
fields = ('category', 'city__province__name', 'city__name',
'name', 'address', 'contact', 'position', 'phone',
'email')
export_order = ('category', 'city__province__name', 'city__name',
'name', 'address', 'contact', 'position', 'phone',
'email')resource의 쓰기 방법은 다음과 같다.
import의 복잡성 때문에 import의resource는 쓰기가 매우 복잡합니다. 왜냐하면 import는 여러 가지 수요가 있기 때문입니다. 예를 들어 어떤 열을 가져왔지만 어떤 열만 업데이트하고 많은 열은 업데이트만 하고 새로 만들지 않으며 열을 가져올 때 여러 가지 데이터 검사...
우선 기초적인 importexport의 InstanceLoader는 매우 중요한 조회 수요를 만족시킬 수 없습니다. 예를 들어 저희 모델에 is 가 있습니다.active 필드에서 이 물건을 내보낼 수 없습니다. 가져올 때 isactive 또 getinstance의 검색 조건그리고 Model Resource 위의 일부 지원도 매우 부족하다. 예를 들어 내가 파일을 입력하면 데이터세트 데이터를 얻을 수 있다. 예를 들어 내가 export를 할 때queryset이 아닌 교체할 수 있는 것을 전달하고 싶고 더 인성화된 오류 힌트를 주고 싶다.
class ModelExtraParamInstanceLoader(BaseInstanceLoader):
""" get_instance , is_active=True"""
def get_queryset(self):
return self.resource._meta.model.objects.all()
def get_instance(self, row):
try:
params = self.resource._meta.import_instanceloader_extra_params
for key in self.resource.get_import_id_fields():
field = self.resource.fields[key]
params[field.attribute] = field.clean(row)
return self.get_queryset().get(**params)
except self.resource._meta.model.DoesNotExist:
return None
class ModelResource(resources.ModelResource):
def set_extra_data(self, extra_data):
self.extra_data = extra_data
def get_clean_row(self, row):
_row = []
for each in row:
if isinstance(each, float):
each = int(each)
each = unicode(each).strip()
_row.append(each)
return _row
def get_dataset_data(self, file_obj):
''' excel '''
headers = self.get_export_headers()
try:
self._dataset_data = get_data_from_excel(file_obj=file_obj, header=headers)
except Exception as ex:
logger.warn(ex)
raise Error(
errors.ExcelFormatError,
err_message=unicode(ex),
message=unicode(ex)
)
return self._dataset_data
def get_printable_row(self, row):
_row = [unicode(each) for each in row]
return u'({})'.format(u', '.join(_row))
def get_printable_error_message(self, error_type, index, row):
return u'excel :[{}]
:{}
:{}'.format(
error_type,
index, self.get_printable_row(row)
)
def get_error(self, error_type, index, row):
return Error(
errors.ExcelFormatError,
err_message='excel ',
message=self.get_printable_error_message(error_type, index, row)
)
def clean_dataset_data(self, data):
''' ， data diff_header
diff_header model , import_data model
， ， raise Error，
'''
headers = self.get_export_headers()
header_length = len(headers)
for index, row in enumerate(data):
if len(row) != header_length:
raise self.get_error(u' ', index+2, row)
return data
def get_dataset(self, file_obj=None):
assert hasattr(self, '_dataset_data') or file_obj, 'You need call get_dataset_data first or pass file_obj'
if file_obj:
data = self.get_dataset_data(file_obj)
else:
data = self._dataset_data
data = self.clean_dataset_data(data)
headers = self.get_diff_headers()
dataset = get_dataset(data, headers)
return dataset
def export(self, queryset=None, empty=False):
"""
Exports a resource.
"""
if queryset is None:
if empty:
if hasattr(self._meta, 'empty_export_data'):
queryset = self._meta.empty_export_data
else:
queryset = []
else:
queryset = self.get_queryset()
headers = self.get_export_headers()
data = tablib.Dataset(headers=headers)
if isinstance(queryset, QuerySet):
# Iterate without the queryset cache, to avoid wasting memory when
# exporting large datasets.
iterable = queryset.iterator()
else:
iterable = queryset
for obj in iterable:
if empty and isinstance(obj, Iterable):
data.append(obj)
else:
data.append(self.export_resource(obj))
return data
def init_instance(self, row=None):
if not row:
row = {}
instance = self._meta.model()
for attr, value in row.iteritems():
setattr(instance, attr, value)
return instance 복잡한 키가 없는 모델의 가져오기 Resource를 보여 줍니다.
class SchoolResource(ModelResource):
def dehydrate_category(self, school):
if school.category == School.MIDDLE_SCHOOL:
return u' '
elif school.category == School.COLLEGE:
return u' '
return ''
def get_export_headers(self):
return [u' ', u' ', u' ', u' ', u' ', u' ',
u' ', u' ', u' ']
def get_diff_headers(self):
return ['category', 'city', 'name', 'address', 'contact', 'position', 'phone', 'email']
def clean_dataset_data(self, data):
data = super(SchoolResource, self).clean_dataset_data(data)
clean_data = []
for index, row in enumerate(data):
_index = index + 2
_row = self.get_clean_row(row)
category = self.clean_dataset_category(_row[0], _index, row)
city = self.clean_dataset_city((_row[1], _row[2]), _index, row)
clean_data.append([category, city]+ _row[3:])
return clean_data
def clean_dataset_category(self, category, index, row):
if category not in (u' ', u' '):
raise self.get_error(u' ', index, row)
if category == u' ':
return 1
else:
return 2
class Meta:
model = School
import_id_fields = ['name',]
import_instanceloader_extra_params = {'is_active': True}
instance_loader_class = ModelExtraParamInstanceLoader
empty_export_data = [...]
fields = ('category', 'city__province__name', 'city__name',
'name', 'address', 'contact', 'position', 'phone',
'email')
export_order = ('category', 'city__province__name', 'city__name',
'name', 'address', 'contact', 'position', 'phone',
'email') resource의 쓰기 방법은 다음과 같다.
class CourseResource(ModelResource):
def dehydrate_is_authentication(self, course):
if course.is_authentication:
return u' '
else:
return u' '
def get_export_headers(self):
return [
u' ', u' ', u' ', u' ',
u' ', u' ', u' ', u' ',
u' '
]
def get_diff_headers(self):
return ['term__name', 'name', 'school__name', 'teacher', 'ID_number', 'phone',
'email', 'is_authentication', 'enrollment']
def init_instance(self, row=None):
if not row:
row = {}
instance = self._meta.model()
for attr, value in row.iteritems():
setattr(instance, attr, value)
instance.term = row['term__name']
instance.school = row['school__name']
return instance
def clean_dataset_data(self, data):
data = super(CourseResource, self).clean_dataset_data(data)
clean_data = []
for index, row in enumerate(data):
_index = index + 2
_row = self.get_clean_row(row)
term = self.clean_dataset_term(_row[0], _index, row)
school = self.clean_dataset_school(_row[2], _index, row)
is_authentication = self.clean_dataset_is_authentication(_row[7], _index, row)
enrollment = self.clean_dataset_enrollment(_row[8], _index, row)
clean_data.append([term, _row[1], school, _row[3], _row[4],
_row[5], _row[6], is_authentication, enrollment])
return clean_data
def clean_dataset_term(self, term, index, row):
try:
return Term.objects.get(name=term, is_active=True)
except Term.DoesNotExist:
raise self.get_error(u' ', index, row)
def clean_dataset_school(self, school, index, row):
try:
school = School.objects.get(name=school, is_active=True)
user = self.extra_data['user']
if not SchoolPermissionFilterBackend().has_school_permission(user,
school):
raise self.get_error(u' ', index, row)
return school
except School.DoesNotExist:
raise self.get_error(u' ', index, row)
def clean_dataset_is_authentication(self, is_authentication, index, row):
if is_authentication == u' ':
return True
if is_authentication == u' ':
return False
raise self.get_error(u' ', index, row)
def clean_dataset_enrollment(self, enrollment, index, row):
try:
if not enrollment:
enrollment = 0
return int(float(enrollment))
except:
raise self.get_error(u' ', index, row)
class Meta:
model = Course
import_id_fields = ['term__name', 'name', 'school__name']
import_instanceloader_extra_params = {
'is_active': True, 'term__is_active': True, 'school__is_active': True}
instance_loader_class = ModelExtraParamInstanceLoader
fields = ('term__name', 'name', 'school__name',
'teacher', 'ID_number', 'phone', 'email', 'is_authentication',
'enrollment')
export_order = ('term__name', 'name', 'school__name',
'teacher', 'ID_number', 'phone', 'email', 'is_authentication',
'enrollment')
empty_export_data = [...] 몇 가지 방법
def extract_data(sheet, header, skip_header=True, row_type='list'):
assert header
data = []
for row_index in xrange(1 if skip_header else 0, sheet.nrows):
row = sheet.row_values(row_index)
assert len(header) == len(row), u'excel {} , '.format(row_index)
if row_type == 'list':
data.append(row)
else:
each_data = {}
for col_index in xrange(len(header)):
each_data[header[col_index]] = row[col_index]
data.append(each_data)
return data
def get_data_from_excel(file_path=None, file_obj=None, header=None,
sheet_index=0, skip_header=True):
assert header
assert file_path or file_obj
if file_path:
with open_workbook(file_path) as wb:
data = extract_data(wb.sheet_by_index(sheet_index), header, skip_header)
else:
with tempinput(file_obj) as tempfilename:
with open_workbook(tempfilename) as wb:
data = extract_data(wb.sheet_by_index(sheet_index), header, skip_header)
return data
def get_dataset(data, header):
return tablib.Dataset(*data, headers=header)
def attachment_response(export_data, filename='download.xls', content_type='application/vnd.ms-excel'):
# Django 1.7 uses the content_type kwarg instead of mimetype
try:
response = HttpResponse(export_data, content_type=content_type)
except TypeError:
response = HttpResponse(export_data, mimetype=content_type)
response['Content-Disposition'] = 'attachment; filename={}'.format(filename)
return response 이 내용에 흥미가 있습니까?
현재 기사가 여러분의 문제를 해결하지 못하는 경우 AI 엔진은 머신러닝 분석(스마트 모델이 방금 만들어져 부정확한 경우가 있을 수 있음)을 통해 가장 유사한 기사를 추천합니다:
Django 시트 업로드우리가 작업한 대부분의 프로젝트에는 일괄적으로 데이터를 업로드하는 기능이 필요했습니다. 우리는 항상 Excel 시트를 처리하고 각 열이 각 모델로 이동하는 다른 코드를 작성했습니다. 최신 프로젝트에서 우리는 그것을 ...
텍스트를 자유롭게 공유하거나 복사할 수 있습니다.하지만 이 문서의 URL은 참조 URL로 남겨 두십시오.
CC BY-SA 2.5, CC BY-SA 3.0 및 CC BY-SA 4.0에 따라 라이센스가 부여됩니다.
좋은 웹페이지 즐겨찾기
