#-*- coding: utf-8 -*-
import json
import logging
import csv
import urllib
from six import StringIO, text_type
import pandas as pd
import numpy as np
from rest_framework.renderers import TemplateHTMLRenderer, JSONRenderer
from rest_framework.parsers import JSONParser, FileUploadParser
from rest_framework.views import APIView
from rest_framework.decorators import api_view, renderer_classes
from rest_framework.response import Response
from rest_framework import status
from rest_framework_csv import renderers as r
from rest_framework_csv.renderers import CSVRenderer
from rest_framework_csv.parsers import CSVParser
from rest_framework.permissions import IsAuthenticated
from kinmatch import (KinmatchException,
AlreadyExistException,
FileTypeException,
DBConnectionException,
NoValueException)
from kinmatch.db import AlleleFrequenciesManager, GenotypeManager, TaskManager
from kinmatch.search import ManageSearcher, ManageCheckBetween, ManageMultipleSearch
from kinmatch.utils import ASTRGenotypeCleaner, YSTRGenotypeCleaner, DataContainer
from kindsite import settings
# six versions 1.3.0 and previous don't have PY2
try:
from six import PY2
except ImportError:
import sys
PY2 = sys.version_info[0] == 2
TEST_CODE = False # TODO : delete me
[문서]class TSVRenderer(CSVRenderer):
media_type = 'text/tsv'
format = 'tsv'
level_sep = '.'
headers = None
[문서] def render(self, data, media_type='text/tsv', renderer_context=None):
if data is None:
return ''
if not isinstance(data, list):
data = [data]
table = self.tablize(data)
csv_buffer = StringIO()
csv_writer = csv.writer(csv_buffer, delimiter='\t')
for row in table:
# Assume that strings should be encoded as UTF-8
csv_writer.writerow([
elem.encode('utf-8') if isinstance(elem, text_type) and PY2 else elem
for elem in row
])
return csv_buffer.getvalue()
[문서]class NotAllowedMethodException(KinmatchException):
def __init__(self, *args, **kwargs):
super(NotAllowedMethodException, self).__init__(*args, **kwargs)
[문서]class BadRequestException(KinmatchException):
def __init__(self, *args, **kwargs):
super(BadRequestException, self).__init__(*args, **kwargs)
PAGINATE_BY = getattr(settings, 'PAGINATE_BY', 10)
log = logging.getLogger(__name__)
[문서]def logger(func):
def caller(*args, **kwargs):
ret = func(*args, **kwargs)
log.info('function name : %s, argurments:%s, %s' %(func.__name__, args, kwargs))
return ret
return caller
[문서]class BaseViewMixin(APIView):
permission_classes = (IsAuthenticated,)
renderer_classes = (JSONRenderer, r.CSVRenderer, TSVRenderer)
def __init__(self, *args, **kwargs):
super(BaseViewMixin, self).__init__(*args, **kwargs)
[문서] def get(self, request, *args, **kwargs):
return self.process(request, action='GET', *args, **kwargs)
[문서] def post(self, request, *args, **kwargs):
return self.process(request, action='POST', *args, **kwargs)
[문서] def put(self, request, *args, **kwargs):
return self.process(request, action='PUT', *args, **kwargs)
[문서] def delete(self, request, *args, **kwargs):
return self.process(request, action='DELETE', *args, **kwargs)
# def process(self, request, format, action, *args, **kwargs):
[문서] def process(self, request, action, *args, **kwargs):
try:
format_ = request.GET.get('format', 'json')
stat = status.HTTP_200_OK
if action == 'POST' or action == 'PUT':
result = self._post(request, format_, *args, **kwargs)
elif action == 'DELETE':
result = self._delete(request, format_, *args, **kwargs)
elif action == 'GET':
result = self._get(request, format_, *args, **kwargs)
else:
pass
except NotAllowedMethodException as e:
log.info('HTTP_405_METHOD_NOT_ALLOWED')
result = DataContainer(status='Exception',
message=str(e),
format_=format_
)
stat = status.HTTP_405_METHOD_NOT_ALLOWED
except AssertionError as e:
log.info('HTTP_500_INTERNAL_SERVER_ERROR')
result = DataContainer(status='Exception',
message=str(e),
format_=format_)
stat = status.HTTP_500_INTERNAL_SERVER_ERROR
except BadRequestException as e:
log.info('HTTP_400_BAD_REQUEST')
result = DataContainer(status='Exception',
message=str(e),
format_=format_)
stat = status.HTTP_400_BAD_REQUEST
except NoValueException as e:
log.info('HTTP_500_INTERNAL_SERVER_ERROR')
result = DataContainer(status='Exception',
message=str(e),
format_=format_)
stat = status.HTTP_500_INTERNAL_SERVER_ERROR
except AlreadyExistException as e:
log.info('HTTP_409_CONFLICT')
result = DataContainer(status='Exception',
message=str(e),
format_=format_)
stat = status.HTTP_409_CONFLICT
except FileTypeException as e:
log.info('HTTP_415_UNSUPPORTED_MEDIA_TYPE')
result = DataContainer(status='Exception',
message=str(e),
format_=format_)
stat = status.HTTP_415_UNSUPPORTED_MEDIA_TYPE
except DBConnectionException as e:
log.info('HTTP_503_SERVICE_UNAVAILABLE')
result = DataContainer(status='Exception',
message=str(e),
format_=format_)
stat = status.HTTP_503_SERVICE_UNAVAILABLE
# except Exception as e:
# log.info('HTTP_500_INTERNAL_SERVER_ERROR')
# result = DataContainer(status='Exception',
# message=str(e),
# format_=format_)
# stat = status.HTTP_500_INTERNAL_SERVER_ERROR
return Response(result.convert(format_), status=stat)
def _post(self, request, format_, *args, **kwargs):
raise NotImplementedError("._post() must be overridden.")
def _put(self, request, format_, *args, **kwargs):
raise NotImplementedError("._put() must be overridden.")
def _get(self, request, format_, *args, **kwargs):
raise NotImplementedError("._get() must be overridden.")
def _delete(self, request, format_, *args, **kwargs):
raise NotImplementedError("._delete() must be overridden.")
def _convert2float(self, value):
return float('inf') if not value else float(value)
def _convert2bool(self, value):
if not value or value.lower() == 'false':
return False
return True
def _convert2none(self, value):
return None if not value else value
def _convert2pageno(self, value):
pageno = 1
try:
pageno = int(value)
except:
pageno = 1
return pageno
def _convert2paginateby(self, value):
paginateby = PAGINATE_BY
if not value:
return paginateby
try:
paginateby = int(value)
except:
paginateby = PAGINATE_BY
return paginateby
[문서]class AlleleFrequenciesMixin(object):
def __init__(self, *args, **kwargs):
super(AlleleFrequenciesMixin, self).__init__(*args, **kwargs)
self.afm = AlleleFrequenciesManager()
[문서] def get_process(self, request, *args, **kwargs):
name = kwargs.get('name', '').strip()
type_ = request.GET.get('type', '').strip()
type_ = self._convert2none(type_)
return self.afm.read(name=name, type_=type_)
[문서] def list_process(self, request, *args, **kwargs):
return self.afm.list()
[문서] def post_process(self, request, *args, **kwargs):
return self.add_or_update(request)
[문서] def put_process(self, request, *args, **kwargs):
return self.add_or_update(request)
[문서] def delete_process(self, request, *args, **kwargs):
name = kwargs.get('name', 'Korean').strip()
return self.afm.delete(name)
[문서] def add_or_update(self, request):
if TEST_CODE:
type_ = request.POST.get('type', 'A-STR').strip()
name = request.POST.get('name', '').strip()
save_by = request.POST.get('save-by', 'new').strip()
informat = request.POST.get('informat', 'CSV').strip()
description = request.POST.get('description', '').strip()
infile = request.FILES['infile']
else:
type_ = request.META.get('HTTP_TYPE', 'A-STR').strip()
name = request.META.get('HTTP_NAME', '').strip()
save_by = request.META.get('HTTP_SAVE_BY', 'new').strip()
informat = request.META.get('HTTP_INFORMAT', 'CSV').strip()
description = request.META.get('HTTP_DESCRIPTION', '').strip()
infile = request.FILES['infile']
description = urllib.parse.unquote(description)
if not type_ in ('A-STR', 'Y-STR', 'SNP'):
raise BadRequestException('type should be one of "A-STR", "Y-STR", "SNP"')
if not informat in ('CSV', 'XLSX'):
raise BadRequestException('informat should be one of "CSV", "XLSX"')
if not save_by in ('new', 'replacement'):
raise BadRequestException('save by option should be one of "new", "replacement"')
if save_by == 'new':
return self.afm.create(infile=infile, informat=informat,
name=name, type_=type_,
description=description)
else:
return self.afm.update(infile=infile, informat=informat,
name=name, type_=type_,
description=description)
[문서]class AlleleFrequenciesView(AlleleFrequenciesMixin, BaseViewMixin):
def __init__(self, *args, **kwargs):
super(AlleleFrequenciesView, self).__init__(*args, **kwargs)
def _get(self, request, format_, *args, **kwargs):
type_ = request.GET.get('type', '').strip()
name = kwargs.get('name', '')
if name:
result = self.get_process(request, *args, **kwargs)
if type_ or format_ == 'json':
return result
else:
return DataContainer(status='Success')
return self.list_process(request, *args, **kwargs)
def _post(self, request, format_, *args, **kwargs):
return self.post_process(request, *args, **kwargs)
def _put(self, request, format_, *args, **kwargs):
return self.put_process(request, *args, **kwargs)
def _delete(self, request, format_, *args, **kwargs):
return self.delete_process(request, *args, **kwargs)
[문서]class GenotypesMixin(object):
def __init__(self, *args, **kwargs):
super(GenotypesMixin, self).__init__(*args, **kwargs)
self.gm = GenotypeManager()
[문서] def get_process(self, request, *args, **kwargs):
type_ = request.GET.get('type', '').strip()
kit = request.GET.get('kit', 'all').strip()
type_ = self._convert2none(type_)
identifier = kwargs.get('identifier', '').strip()
kit_temp = {'all':'All', 'ab':'AB', 'Promega':'Promega'}
kit = kit_temp.get(kit, 'All')
return self.gm.read(identifier=identifier, type_=type_, kit=kit)
[문서] def list_process(self, request, *args, **kwargs):
type_ = request.GET.get('type', '').strip()
identifier = request.GET.get('identifier', '').strip()
page_no = request.GET.get('page-no', '').strip()
paginate_by = request.GET.get('paginate-by', '').strip()
sort = request.GET.get('sort', '').strip()
desc = request.GET.get('desc', '').strip()
filter_astr = request.GET.get('filter-astr', '').strip()
filter_ystr = request.GET.get('filter-ystr', '').strip()
filter_mtdna = request.GET.get('filter-mtdna', '').strip()
if type_ and not sort != 'identifier':
raise BadRequestException("type parameter does not work with sorting identifier")
type_ = self._convert2none(type_)
page_no = self._convert2pageno(page_no)
paginate_by = self._convert2paginateby(paginate_by)
sort = self._convert2none(sort)
desc = self._convert2bool(desc)
filter_astr = self._convert2bool(filter_astr)
filter_ystr = self._convert2bool(filter_ystr)
filter_mtdna = self._convert2bool(filter_mtdna)
genotypes = self.gm.list(type_=type_,
identifier=identifier,
page=page_no,
paginate_by=paginate_by,
sort=sort,
desc=desc,
filter_astr=filter_astr,
filter_ystr=filter_ystr,
filter_mtdna=filter_mtdna
)
return genotypes
[문서] def post_process(self, request, *args, **kwargs):
return self.add(request)
[문서] def put_process(self, request, *args, **kwargs):
return self.add(request)
[문서] def delete_process(self, request, *args, **kwargs):
identifier = kwargs.get('identifier', '').strip()
genotype = kwargs.get('genotype', 'all').strip()
if not identifier:
raise BadRequestException('identifier is needed to delete')
return self.gm.delete(identifier, genotype)
[문서] def add(self, request):
infile = request.FILES.get('infile', '')
if infile:
return self.add_or_update_genotypefile(request)
else:
return self.update_genotype(request)
[문서] def add_or_update_genotypefile(self, request):
if TEST_CODE:
type_ = request.POST.get('type', 'A-STR').strip()
informat = request.POST.get('informat', 'GeneMark').strip()
group = request.POST.get('group', '').strip()
save_by = request.POST.get('save-by', 'replacement').strip()
infile = request.FILES['infile']
else:
type_ = request.META.get('HTTP_TYPE', 'A-STR').strip()
informat = request.META.get('HTTP_INFORMAT', 'GeneMark').strip()
group = request.META.get('HTTP_GROUP', '').strip()
save_by = request.META.get('HTTP_SAVE_BY', 'replacement').strip()
infile = request.FILES['infile']
if not informat in ('GeneMark', 'JSON', 'CSV', 'XLSX'):
raise BadRequestException('Input file format should be one of "GeneMark", "JSON", "CSV" or "XLSX"')
if not type_ in ('A-STR', 'Y-STR', 'SNP', 'mtDNA'):
raise BadRequestException('type should be one of "A-STR", "Y-STR", "SNP" or "mtDNA"')
if not save_by in ('new', 'replacement', 'merge-with-overwrite', 'merge-without-overwrite'):
raise BadRequestException('save by option should be one of "new", "replacement", "merge-with-overwrite", "merge-without-overwrite"')
group = self._convert2none(group)
if save_by == 'new':
return self.gm.create(infile=infile,
informat=informat,
type_=type_,
group=group)
return self.gm.update(file_or_individual='file',
infile=infile,
informat=informat,
type_=type_,
save_by=save_by,
group=group)
[문서] def update_genotype(self, request):
type_ = request.DATA.get('type', '')
identifier = request.DATA.get('identifier', '')
genotype = request.DATA.get('genotype', '')
if not type_ in ('A-STR', 'Y-STR', 'SNP', 'mtDNA'):
raise BadRequestException('type should be one of "A-STR", "Y-STR", "SNP" or "mtDNA"')
if not identifier:
raise BadRequestException('identifier is not valid.')
return self.gm.update(file_or_individual='individual',
type_=type_,
identifier=identifier,
genotype=genotype)
[문서]class GenotypesView(GenotypesMixin, BaseViewMixin):
def __init__(self, *args, **kwargs):
super(GenotypesView, self).__init__(*args, **kwargs)
def _get(self, request, format_, *args, **kwargs):
type_ = request.GET.get('type', '').strip()
type_ = self._convert2none(type_)
identifier = kwargs.get('identifier', '').strip()
if identifier:
result = self.get_process(request, *args, **kwargs)
if type_ or format_ == 'json':
return result
else:
return DataContainer(status='Success')
return self.list_process(request, *args, **kwargs)
def _post(self, request, format_, *args, **kwargs):
return self.post_process(request, *args, **kwargs)
def _put(self, request, format_, *args, **kwargs):
return self.put_process(request, *args, **kwargs)
def _delete(self, request, format_, *args, **kwargs):
return self.delete_process(request, *args, **kwargs)
[문서]class GroupMixin(object):
def __init__(self, *args, **kwargs):
super(GroupMixin, self).__init__(*args, **kwargs)
[문서] def get_process(self, request, *args, **kwargs):
return None
[문서] def list_process(self, request, *args, **kwargs):
return None
[문서] def post_process(self, request, *args, **kwargs):
return None
[문서] def put_process(self, request, *args, **kwargs):
return None
[문서] def delete_process(self, request, *args, **kwargs):
return None
[문서]class GroupView(GroupMixin, BaseViewMixin):
def __init__(self, *args, **kwargs):
super(GroupView, self).__init__(*args, **kwargs)
def _get(self, request, format_, *args, **kwargs):
return self.list_process(request, *args, **kwargs)
def _post(self, request, format_, *args, **kwargs):
return self.post_process(request, *args, **kwargs)
def _put(self, request, format_, *args, **kwargs):
return self.put_process(request, *args, **kwargs)
def _delete(self, request, format_, *args, **kwargs):
return self.delete_process(request, *args, **kwargs)
[문서]class TaskMixin(object):
def __init__(self, *args, **kwargs):
super(TaskMixin, self).__init__(*args, **kwargs)
self.tm = TaskManager()
[문서] def get_process(self, request, *args, **kwargs):
identifier = kwargs.get('identifier', '')
return self.tm.read(identifier)
[문서] def list_process(self, request, *args, **kwargs):
identifier = request.GET.get('identifier', '')
query_group = request.GET.get('query-group', '')
query_type = request.GET.get('query-type', '')
target_group = request.GET.get('target-group', '')
target_type = request.GET.get('target-type', '')
description = request.GET.get('description', '')
status = request.GET.get('status', '')
sort = request.GET.get('sort', 'registered_time')
desc = request.GET.get('desc', 'false')
paginate_by = request.GET.get('paginate-by', '20')
page = request.GET.get('page-no', '1')
# if not query_type in ['all', 'range', 'keyword', 'manual']:
# raise BadRequestException('relationship should be one of \
# "all", "range", "manual" or "keyword"')
# if not target_type in ['all', 'range', 'keyword', 'manual']:
# raise BadRequestException('relationship should be one of \
# "all", "range", "manual" or "keyword"')
# if not sort in ['identifier', 'query_group', 'query_type',
# 'query_count', 'target_group', 'target_type',
# 'target_count', 'status', 'progress',
# 'registered_time']:
# raise BadRequestException('sort should be one of \
# "identifier", "query_group", "query_type",\
# "query_count", "target_group", "target_type",\
# "target_count", "status", "progress" or \
# "registered_time"')
identifier = self._convert2none(identifier)
query_group = self._convert2none(query_group)
query_type = self._convert2none(query_type)
target_group = self._convert2none(target_group)
target_type = self._convert2none(target_type)
description = self._convert2none(description)
status = self._convert2none(status)
desc = self._convert2bool(desc)
paginate_by = self._convert2paginateby(paginate_by)
page = self._convert2pageno(page)
return self.tm.list(identifier=identifier,
query_group=query_group,
query_type=query_type,
target_group=target_group,
target_type=target_type,
description=description,
status=status,
sort=sort,
desc=desc,
paginate_by=paginate_by,
page=page)
[문서] def post_process(self, request, *args, **kwargs):
return None
[문서] def put_process(self, request, *args, **kwargs):
return None
[문서] def delete_process(self, request, *args, **kwargs):
identifier = kwargs.get('identifier', '')
return self.tm.delete(identifier=identifier)
[문서]class TaskView(TaskMixin, BaseViewMixin):
def __init__(self, *args, **kwargs):
super(TaskView, self).__init__(*args, **kwargs)
def _get(self, request, format_, *args, **kwargs):
identifier = kwargs.get('identifier', '')
if identifier:
return self.get_process(request, *args, **kwargs)
return self.list_process(request, *args, **kwargs)
def _post(self, request, format_, *args, **kwargs):
raise NotAllowedMethodException('not allowed method exception')
def _put(self, request, format_, *args, **kwargs):
raise NotAllowedMethodException('not allowed method exception')
def _delete(self, request, format_, *args, **kwargs):
return self.delete_process(request, *args, **kwargs)
[문서]class SearchMixin(object):
def __init__(self, *args, **kwargs):
super(SearchMixin, self).__init__(*args, **kwargs)
[문서] def search(self, request, *args, **kwargs):
identifier = request.GET.get('identifier', '').strip()
target_group = request.GET.get('target-group', '').strip()
relationship = request.GET.get('relationship', '').strip()
partial = request.GET.get('partial', 'true').strip()
check_astr = request.GET.get('check-astr', 'true').strip()
astr_partial = request.GET.get('astr-partial', 'true').strip()
astr_permit_mismatches = request.GET.get('astr-permit-mismatches', '2').strip()
af_table = request.GET.get('af-table', '').strip()
min_kinship_index = request.GET.get('min-kinship-index', '1').strip()
check_ystr = request.GET.get('check-ystr', 'false').strip()
ystr_partial = request.GET.get('ystr-partial', 'false').strip()
ystr_permit_mismatches = request.GET.get('ystr-permit-mismatches', '2').strip()
check_mtdna = request.GET.get('check-mtdna', 'false').strip()
mtdna_partial = request.GET.get('mtdna-partial', 'false').strip()
mtdna_permit_mismatches = request.GET.get('mtdna-permit-mismatches', '2').strip()
mtdna_exclude_cstretches = request.GET.get('mtdna-exclude-cstretches', 'false').strip()
dbname = request.GET.get('dbname', 'kinmatch').strip()
format_ = request.GET.get('format', 'json').strip()
if not relationship in ('identity', 'paternity', 'kinship'):
raise BadRequestException('relationship should be one of "identity", "paternity" or "kinship"')
if not identifier:
raise BadRequestException('identifier is needed to search')
if not target_group:
raise BadRequestException('target_group is needed to search')
partial = self._convert2bool(partial)
check_astr = self._convert2bool(check_astr)
astr_partial = self._convert2bool(astr_partial)
astr_permit_mismatches = self._convert2float(astr_permit_mismatches)
af_table = self._convert2none(af_table)
min_kinship_index = self._convert2float(min_kinship_index)
check_ystr = self._convert2bool(check_ystr)
ystr_partial = self._convert2bool(ystr_partial)
ystr_permit_mismatches = self._convert2float(ystr_permit_mismatches)
check_mtdna = self._convert2bool(check_mtdna)
mtdna_partial = self._convert2bool(mtdna_partial)
mtdna_permit_mismatches = self._convert2float(mtdna_permit_mismatches)
mtdna_exclude_cstretches = self._convert2bool(mtdna_exclude_cstretches)
ns = ManageSearcher(relationship_type=relationship,
identifier=identifier,
target_group=target_group,
partial=partial,
check_astr=check_astr,
astr_partial=astr_partial,
astr_permit_mismatches=astr_permit_mismatches,
af_table=af_table,
min_kinship_index=min_kinship_index,
check_ystr=check_ystr,
ystr_partial=ystr_partial,
ystr_permit_mismatches=ystr_permit_mismatches,
check_mtdna=check_mtdna,
mtdna_partial=mtdna_partial,
mtdna_permit_mismatches=mtdna_permit_mismatches,
mtdna_exclude_cstretches=mtdna_exclude_cstretches,
dbname=dbname
)
return ns.search()
[문서] def search_multiple(self, request, *args, **kwargs):
query_group = request.GET.get('query-group', 'soldier').strip()
query_type = request.GET.get('query-type', 'all').strip()
query_first = request.GET.get('query-first', '').strip()
query_second = request.GET.get('query-second', '').strip()
target_group = request.GET.get('target-group', 'customer').strip()
target_type = request.GET.get('target-type', 'all').strip()
target_first = request.GET.get('target-first', '').strip()
target_second = request.GET.get('target-second', '').strip()
relationship = request.GET.get('relationship', '').strip()
partial = request.GET.get('partial', 'true').strip()
check_astr = request.GET.get('check-astr', 'true').strip()
astr_partial = request.GET.get('astr-partial', 'true').strip()
astr_permit_mismatches = request.GET.get('astr-permit-mismatches', '2').strip()
af_table = request.GET.get('af-table', '').strip()
min_kinship_index = request.GET.get('min-kinship-index', '1').strip()
check_ystr = request.GET.get('check-ystr', 'false').strip()
ystr_partial = request.GET.get('ystr-partial', 'false').strip()
ystr_permit_mismatches = request.GET.get('ystr-permit-mismatches', '2').strip()
check_mtdna = request.GET.get('check-mtdna', 'false').strip()
mtdna_partial = request.GET.get('mtdna-partial', 'false').strip()
mtdna_permit_mismatches = request.GET.get('mtdna-permit-mismatches',
'2').strip()
mtdna_exclude_cstretches = request.GET.get('mtdna-exclude-cstretches',
'false').strip()
dbname = request.GET.get('dbname', 'kinmatch').strip()
description = request.GET.get('description', '').strip()
format_ = request.GET.get('format', 'json').strip()
if query_type not in ['all', 'range', 'keyword', 'manual']:
raise BadRequestException(
'{} is not proper value for query_type'.format(query_type))
if target_type not in ['all', 'range', 'keyword', 'manual']:
raise ParameterException(
'{} is not proper value for target_type'.format(target_type))
query_first = self._convert2none(query_first)
query_second = self._convert2none(query_second)
target_first = self._convert2none(target_first)
target_second = self._convert2none(target_second)
partial = self._convert2bool(partial)
check_astr = self._convert2bool(check_astr)
astr_partial = self._convert2bool(astr_partial)
astr_permit_mismatches = self._convert2float(astr_permit_mismatches)
af_table = self._convert2none(af_table)
min_kinship_index = self._convert2float(min_kinship_index)
check_ystr = self._convert2bool(check_ystr)
ystr_partial = self._convert2bool(ystr_partial)
ystr_permit_mismatches = self._convert2float(ystr_permit_mismatches)
check_mtdna = self._convert2bool(check_mtdna)
mtdna_partial = self._convert2bool(mtdna_partial)
mtdna_permit_mismatches = self._convert2float(mtdna_permit_mismatches)
mtdna_exclude_cstretches = self._convert2bool(mtdna_exclude_cstretches)
mms = ManageMultipleSearch(query_group=query_group,
query_type=query_type,
query_first=query_first,
query_second=query_second,
target_group=target_group,
target_type=target_type,
target_first=target_first,
target_second=target_second,
relationship_type=relationship,
partial=partial,
check_astr=check_astr,
astr_partial=astr_partial,
astr_permit_mismatches=astr_permit_mismatches,
af_table=af_table,
min_kinship_index=min_kinship_index,
check_ystr=check_ystr,
ystr_partial=ystr_partial,
ystr_permit_mismatches=ystr_permit_mismatches,
check_mtdna=check_mtdna,
mtdna_partial=mtdna_partial,
mtdna_permit_mismatches=mtdna_permit_mismatches,
mtdna_exclude_cstretches=mtdna_exclude_cstretches,
dbname=dbname
)
return mms.search()
[문서] def check_between(self, request, *args, **kwargs):
identifier_a = request.GET.get('identifier-a', '').strip()
identifier_b = request.GET.get('identifier-b', '').strip()
relationship = request.GET.get('relationship', '').strip()
partial = request.GET.get('partial', 'true').strip()
check_astr = request.GET.get('check-astr', 'true').strip()
astr_partial = request.GET.get('astr-partial', 'true').strip()
astr_permit_mismatches = request.GET.get('astr-permit-mismatches', '2').strip()
af_table = request.GET.get('af-table', '').strip()
min_kinship_index = request.GET.get('min-kinship-index', '1').strip()
check_ystr = request.GET.get('check-ystr', 'false').strip()
ystr_partial = request.GET.get('ystr-partial', 'false').strip()
ystr_permit_mismatches = request.GET.get('ystr-permit-mismatches', '2').strip()
check_mtdna = request.GET.get('check-mtdna', 'false').strip()
mtdna_partial = request.GET.get('mtdna-partial', 'false').strip()
mtdna_permit_mismatches = request.GET.get('mtdna-permit-mismatches', '2').strip()
mtdna_exclude_cstretches = request.GET.get('mtdna-exclude-cstretches', 'false').strip()
dbname = request.GET.get('dbname', 'kinmatch').strip()
format_ = request.GET.get('format', 'json').strip()
if not relationship in ('identity', 'paternity', 'kinship'):
raise BadRequestException('relationship should be one of "identity", "paternity" or "kinship"')
if not identifier_a:
raise BadRequestException('identifier a is needed to search')
if not identifier_b:
raise BadRequestException('identifier b is needed to search')
if identifier_a == identifier_b:
raise BadRequestException('identifier a and identifier b are the same')
partial = self._convert2bool(partial)
check_astr = self._convert2bool(check_astr)
astr_partial = self._convert2bool(astr_partial)
astr_permit_mismatches = self._convert2float(astr_permit_mismatches)
af_table = self._convert2none(af_table)
min_kinship_index = self._convert2float(min_kinship_index)
check_ystr = self._convert2bool(check_ystr)
ystr_partial = self._convert2bool(ystr_partial)
ystr_permit_mismatches = self._convert2float(ystr_permit_mismatches)
check_mtdna = self._convert2bool(check_mtdna)
mtdna_partial = self._convert2bool(mtdna_partial)
mtdna_permit_mismatches = self._convert2float(mtdna_permit_mismatches)
mtdna_exclude_cstretches = self._convert2bool(mtdna_exclude_cstretches)
mcb = ManageCheckBetween(relationship_type=relationship,
identifier_a=identifier_a,
identifier_b=identifier_b,
partial=partial,
check_astr=check_astr,
astr_partial=astr_partial,
astr_permit_mismatches=astr_permit_mismatches,
af_table=af_table,
min_kinship_index=min_kinship_index,
check_ystr=check_ystr,
ystr_partial=ystr_partial,
ystr_permit_mismatches=ystr_permit_mismatches,
check_mtdna=check_mtdna,
mtdna_partial=mtdna_partial,
mtdna_permit_mismatches=mtdna_permit_mismatches,
mtdna_exclude_cstretches=mtdna_exclude_cstretches,
dbname=dbname
)
return mcb.check()
[문서]class SearchBaseViewMixin(BaseViewMixin):
def __init__(self, *args, **kwargs):
super(SearchBaseViewMixin, self).__init__(*args, **kwargs)
def _get(self, request, *args, **kwargs):
NotImplementedError("._get() must be overridden.")
def _post(self, request, *args, **kwargs):
raise NotAllowedMethodException('not allowed method exception')
def _put(self, request, *args, **kwargs):
raise NotAllowedMethodException('not allowed method exception')
def _delete(self, request, *args, **kwargs):
raise NotAllowedMethodException('not allowed method exception')
[문서]class SearchView(SearchMixin, SearchBaseViewMixin):
def __init__(self, *args, **kwargs):
super(SearchView, self).__init__(*args, **kwargs)
def _get(self, request, *args, **kwargs):
return self.search(request, *args, **kwargs)
[문서]class SearchMultipleView(SearchMixin, SearchBaseViewMixin):
def __init__(self, *args, **kwargs):
super(SearchMultipleView, self).__init__(*args, **kwargs)
def _get(self, request, *args, **kwargs):
return self.search_multiple(request, *args, **kwargs)
[문서]class CheckBetweenView(SearchMixin, SearchBaseViewMixin):
def __init__(self, *args, **kwargs):
super(CheckBetweenView, self).__init__(*args, **kwargs)
def _get(self, request, *args, **kwargs):
return self.check_between(request, *args, **kwargs)