Commit 8c4708de authored by Lucas Gueiros's avatar Lucas Gueiros
Browse files

copiando arquivos

parents
# CAUTION: PYTHON VODOO AHEAD!
## Utilização
Esse módulo pode ser utilizado para consumir (apenas função read) a API dos SIGs.
### Configuração
Instale o módulo...
### Configurações
No settings do seu projeto, adicione as seguintes propriedades:
* SIGS_API_BASE_URL = url de consulta da API
* SIGS_API_AUTHENTICATION_URL = url de autenticação da API
* SIGS_API_KEY = chave de API
* SIGS_API_CLIENT_ID = usuário
* SIGS_API_CLIENT_SECRET = senha
### No seu arquivo models
Importe:
```python
from api_sigs.models import SIGsAPIField, SIGsAPIModel, SIGsAPIForeignKey
```
Defina as classes:
```python
class UnidadeFederativaModel(SIGsAPIModel):
descricao = SIGsAPIField()
sigla = SIGsAPIField()
id_pais = SIGsAPIField()
id_unidade_federativa = SIGsAPIField()
class Meta:
servico = "comum"
recurso = "unidades-federativas"
class MunicipioModel(SIGsAPIModel):
nome_municipio = SIGsAPIField()
id_municipio = SIGsAPIField()
unidade_federativa = SIGsAPIForeignKey(UnidadeFederativaModel)
class Meta:
servico = "comum"
recurso = "municipios"
```
# ideias
Ideias de apis para consumir:
calendario da universidade (feriados etc.)
calendário acadêmico
livros da biblioteca, trabalhos de conclusão de curso
editais de concursos
cursos, componentes curriculares
eventos
telefones - ramais
from django.apps import AppConfig
class ApiSigsConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'api_sigs'
_missing = object()
class cached_property(object):
"""A decorator that converts a function into a lazy property. The
function wrapped is called the first time to retrieve the result
and then that calculated result is used the next time you access
the value::
class Foo(object):
@cached_property
def foo(self):
# calculate something important here
return 42
The class has to have a `__dict__` in order for this property to
work.
"""
# implementation detail: this property is implemented as non-data
# descriptor. non-data descriptors are only invoked if there is
# no entry with the same name in the instance's __dict__.
# this allows us to completely get rid of the access function call
# overhead. If one choses to invoke __get__ by hand the property
# will still work as expected because the lookup logic is replicated
# in __get__ for manual invocation.
def __init__(self, func, name=None, doc=None):
self.__name__ = name or func.__name__
self.__module__ = func.__module__
self.__doc__ = doc or func.__doc__
self.func = func
def __get__(self, obj, type=None):
if obj is None:
return self
value = obj.__dict__.get(self.__name__, _missing)
if value is _missing:
value = self.func(obj)
obj.__dict__[self.__name__] = value
return value
\ No newline at end of file
from .basic import SIGsAPIField, SIGsAPIDateTimeField, SIGsAPIIdField, SIGsAPIStringField
from .relation import SIGsAPIDetailsField, SIGsAPIForeignKey, SIGSAPIOneToManyField, SIGsAPIManyToOne, SIGsAPIOneToOneField, SIGsToOneRelation, SIGsToManyRelation
\ No newline at end of file
from .basic import SIGsAPIField
from django.utils.functional import cached_property
from ..models import SIGsAPIRelationModelManager
from django.db.models import IntegerField, Field, CharField
import importlib
from django.forms import ChoiceField, ModelChoiceField
from django.forms.widgets import Select
def add_to_one (
cls,
target,
attribute,
origin_key,
target_key,
field,
):
# Verifique se já existe um accessor para o atributo direto
if not hasattr(cls, origin_key):
cls = SIGsAPIField.contribute_to_sigs_api_class(
field,
cls,
origin_key,
)
cls._meta.set_attribute(attribute, field)
func = cached_property( lambda _self: target.objects.get(**{target_key: getattr(_self, origin_key)}))
func.__set_name__(cls, attribute)
setattr(cls, attribute, func)
def add_to_many (cls,
target,
attribute,
origin_key,
target_key,
field,
):
cls._meta.set_attribute(attribute, field) # Tem certeza que é o mesmo?
related_manager_func = cached_property( lambda _self: SIGsAPIRelationModelManager(target,{
target_key: getattr(_self, origin_key),
}))
related_manager_func.__set_name__(cls, attribute)
setattr(cls, attribute, related_manager_func)
class SIGsToManyRelation(SIGsAPIField):
def __init__(self,
target_class,
target_class_relation,# = None,
origin_class_fk = "pk",
target_class_fk = "pk",
mapped_by_target = False,
null = False,
):
self.target_class = target_class
self.origin_class_fk = origin_class_fk
self.target_class_fk = target_class_fk
self.target_class_relation = target_class_relation or None
self.mapped_by_target = mapped_by_target
super().__init__(property_name= self.origin_class_fk,
null = null)
def inverted(self):
inverted = SIGsToManyRelation(
target_class= self.origin_class,
target_class_relation = self.origin_class_relation,
target_class_fk = self.origin_class_fk,
origin_class_fk= self.target_class_fk,
mapped_by_target= not self.mapped_by_target,
)
inverted.origin_class = self.target_class
inverted.origin_class_relation = self.origin_class_relation
inverted.opposite_field = self
self.opposite_field = inverted
return inverted
class SIGsToOneRelation(SIGsAPIField):
def __init__(self,
target_class,
target_class_relation,# = None,
origin_class_fk = "pk",
target_class_fk = "pk",
mapped_by_target = False,
):
self.target_class = target_class
self.origin_class_fk = origin_class_fk
self.target_class_fk = target_class_fk
self.target_class_relation = target_class_relation or None
self.mapped_by_target = mapped_by_target
super().__init__(property_name= self.origin_class_fk,
null = False)
def inverted(self):
inverted = SIGsToManyRelation(
target_class= self.origin_class,
target_class_relation = self.origin_class_relation,
target_class_fk = self.origin_class_fk,
origin_class_fk= self.target_class_fk,
mapped_by_target= not self.mapped_by_target,
)
inverted.origin_class = self.target_class
inverted.origin_class_relation = self.origin_class_relation
inverted.opposite_field = self
self.opposite_field = inverted
return inverted
def contribute_to_sigs_api_class(self,
origin_class,
origin_class_relation):
self.origin_class = origin_class
self.origin_class_relation = origin_class_relation
self.opposite_field = self.inverted()
add_to_one (cls = origin_class,
target = self.target_class,
attribute = origin_class_relation,
origin_key = self.origin_class_fk,
target_key = self.target_class_fk,
field = self,
)
if self.origin_class_fk == "pk":
self.attribute_name = origin_class_relation
return origin_class
return super().contribute_to_sigs_api_class(origin_class, self.origin_class_fk.replace('-','_'))
class SIGsAPIManyToOne(SIGsToOneRelation):
def __init__(self,
target_class,
target_class_relation,
origin_class_fk = "pk",
target_class_fk = "pk",
mapped_by_target = False,
null = False,
):
super().__init__(
target_class=target_class,
target_class_relation=target_class_relation,
origin_class_fk= origin_class_fk,
target_class_fk = target_class_fk,
mapped_by_target= mapped_by_target,
)
def contribute_to_sigs_api_class(self,
origin_class,
origin_class_relation):
origin_class = super().contribute_to_sigs_api_class(origin_class, origin_class_relation)
add_to_many(cls = self.target_class,
target = origin_class,
attribute=self.target_class_relation,
origin_key = self.target_class_fk,
target_key=self.origin_class_fk,
field = self.opposite_field,
)
return origin_class
class SIGSAPIOneToManyField (SIGsAPIField):
def __init__(self,
target_class,
target_class_relation,# = None,
target_class_fk = "pk",
):
self.target_class = target_class
self.target_class_fk = target_class_fk
self.target_class_relation = target_class_relation or None
super().__init__()
def contribute_to_sigs_api_class(self,
origin_class,
origin_class_relation):
self.attribute_name = origin_class_relation
#if not self.related_name:
# self.related_name = re.sub('([A-Z]{1})',r'-\1',cls.__name__)[1:].replace('-Model','s').lower()
# O nome de propriedade normalmente a gente tira o "id-", então eu o coloco de volta
#property_name_on_object = "id-" + property_name.replace('_','-')
#if not self.related_name_pk:
# self.related_name_pk = "id_" + property_name
# Vamos setar a propriedade na OUTRA classe
#self.target_class._meta.set_attribute(self.target_class_relation, self)
# Adicionando FK
func_fk = cached_property( lambda _self: _self._object[self.target_class_fk.replace('_','-')])
func_fk.__set_name__(self.target_class, self.target_class_fk)
setattr(self.target_class, self.target_class_fk, func_fk)
# Adicionando a relação
func = cached_property( lambda _self: origin_class.objects.get(**{self.origin_class_fk: getattr(_self, self.target_class_fk)}))
func.__set_name__(self.target_class, self.target_class_relation)
setattr(self.target_class, self.target_class_relation, func)
# Agora vamos colocar a propriedade na classe origem
origin_class._meta.set_attribute(origin_class_relation, self) # Tem certeza que é o mesmo?
target_manager_func = cached_property( lambda _self: SIGsAPIRelationModelManager(self.target_class, filters = {
self.target_class_fk: _self.pk,
}))
target_manager_func.__set_name__(origin_class, origin_class_relation, )
setattr(origin_class, origin_class_relation, target_manager_func)
return origin_class
class SIGsAPIDetailsField (SIGsAPIField):
def __init__(self,
target_class,
origin_fk = "pk",
target_fk = "pk",
target_class_relation = None,# = None,
create_on_target = True,
create_on_origin = True,
):
self.target_class = target_class
self.target_class_fk = target_fk
self.origin_class_fk = origin_fk
self.target_class_relation = target_class_relation or None
self.create_on_target = create_on_target
self.create_on_origin = create_on_origin
super().__init__()
def contribute_to_sigs_api_class(self,
origin_class,
origin_class_relation):
self.attribute_name = origin_class_relation
#if not self.related_name:
# self.related_name = re.sub('([A-Z]{1})',r'-\1',cls.__name__)[1:].replace('-Model','s').lower()
# O nome de propriedade normalmente a gente tira o "id-", então eu o coloco de volta
#property_name_on_object = "id-" + property_name.replace('_','-')
#if not self.related_name_pk:
# self.related_name_pk = "id_" + property_name
# Vamos setar a propriedade na própria classe
if self.create_on_origin:
origin_class._meta.set_attribute(origin_class_relation, self)
func = cached_property( lambda _self: self.target_class.objects.get(pk = getattr(_self, self.origin_class_fk)))
func.__set_name__(origin_class, origin_class_relation)
setattr(origin_class, origin_class_relation, func)
if self.origin_class_fk != 'pk':
origin_class = super().contribute_to_sigs_api_class(origin_class, self.origin_class_fk.replace('-','_'))
if self.create_on_target:
# Adicionando FK, mas apenas se não for também PK
if self.target_class_fk != 'pk':
func_fk = cached_property( lambda _self: _self._object[self.target_class_fk.replace('_','-')])
func_fk.__set_name__(self.target_class, self.target_class_fk)
setattr(self.target_class, self.target_class_fk, func_fk)
# Adicionando a relação
func = cached_property( lambda _self: origin_class.objects.get(**{self.origin_class_fk: getattr(_self, self.target_class_fk)}))
func.__set_name__(self.target_class, self.target_class_relation)
setattr(self.target_class, self.target_class_relation, func)
return origin_class
class SIGsAPIOneToOneField (SIGsToOneRelation):
def __init__(self,
target_class,
target_class_relation,
origin_class_fk = "pk",
target_class_fk = "pk",
mapped_by_target = False,
null = False,
):
super().__init__(
target_class=target_class,
target_class_relation=target_class_relation,
origin_class_fk= origin_class_fk,
target_class_fk = target_class_fk,
mapped_by_target= mapped_by_target,
)
def contribute_to_sigs_api_class(self,
origin_class,
origin_class_relation):
origin_class = super().contribute_to_sigs_api_class(origin_class, origin_class_relation)
add_to_one(cls = self.target_class,
target = origin_class,
attribute=self.target_class_relation,
origin_key = self.target_class_fk,
target_key=self.origin_class_fk,
field = self.opposite_field,
)
return origin_class
class SingleRelatedAccessor:
def __init__(self, field):
self.field = field
def __get__(self, instance, cls=None):
"""Recupera o objeto pela API SIGs usando o ID salvo no banco
Por exemplo, o campo uf_unidade_federativa
Args:
intance (_type_): é o objeto!
cls (_type_, optional): é a classe do objeto. Defaults to None.
"""
return self.field.target_class.objects.get(**{
self.field.target_class_fk: getattr(instance, self.field.get_attname())
})
def __set__(self, instance, value):
"""
Altera a entitdade relacionada.
Quando alguem fizer ``child.parent = parent``:
- ``self`` is the descriptor managing the ``parent`` attribute
- ``instance`` is the ``child`` instance
- ``value`` is the ``parent`` instance on the right of the equal sign
"""
if value is None:
if self.field.null:
setattr(instance, self.field.get_attname(), None)
else:
raise ValueError("O valor não pode ser nulo")
elif isinstance(value, str):
setattr(instance, self.field.get_attname(), value)
elif not isinstance(value, self.field.target_class):
raise TypeError("Objeto {o}, do tipo {t}, não é da classe {c} dessa relação".format(o = value, t = value.__class__, c = self.field.target_class))
else:
setattr(instance, self.field.get_attname(), getattr(value, self.field.target_class_fk))
class SIGsAPIForeignKeyFormField (ChoiceField):
def __init__(self, **kwargs):
super().__init__(choices = kwargs['choices_form_class'].get_sigsapi_choices, **kwargs)
class SIGsAPIForeignKey(CharField, SIGsAPIManyToOne):
"""Guarda referência para um entidade na API dos SIGs"""
def __init__(self,
target_class,
target_class_relation = None,
origin_class_fk = "pk",
target_class_fk = "pk",
*args,
**kwargs):
SIGsAPIManyToOne.__init__(
self,
target_class = target_class,
target_class_relation = target_class_relation,
origin_class_fk = origin_class_fk,
target_class_fk = target_class_fk,
)
default = kwargs.pop('default')
if(isinstance(default, target_class)):
kwargs['default'] = getattr(default, self.target_class_fk)
else:
kwargs['default'] = default
kwargs['max_length'] = 100
#kwargs['choices'] = self.get_sigsapi_choices()
CharField.__init__(self, *args, **kwargs)
@property
def non_db_attrs(self):
return super().non_db_attrs + ("target_class","target_class_relation","origin_class_fk","target_class_fk")
def deconstruct(self):
name, path, args, kwargs = super().deconstruct()
kwargs['target_class'] = self.target_class
return name, path, args, kwargs
def get_attname(self):
return "%s_id" % self.name
def get_attname_column(self):
attname = self.get_attname()
column = self.db_column or attname
return attname, column
def contribute_to_class(self, cls, name, private_only=False, **kwargs):
# ADicione o campo simples, integer, que vai salvar o ID
IntegerField.contribute_to_class(self, cls, name, private_only=private_only, **kwargs)
# Adicione o Accessor, que vai permitir acessar diretamente o objeto
setattr(cls, name, SingleRelatedAccessor(self))
if isinstance(self.target_class, str):
splitted = self.target_class.split('.')
class_name = splitted[-1]
module_name = '.'.join(splitted[0:-1])
module = importlib.import_module(module_name)
self.target_class = getattr(module, class_name)
if not self.target_class_relation:
self.target_class_relation = cls.__name__.lower()
# Agora vamos colocar a propriedade na classe target
self.target_class._meta.set_attribute(self.target_class_relation, self) # Tem certeza que é o mesmo?
related_manager_func = cached_property( lambda _self: SIGsAPIRelationModelManager(cls,{
self.origin_class_fk: _self.pk
}))
related_manager_func.__set_name__(self.target_class, self.target_class_relation)
setattr(self.target_class, self.target_class_relation, related_manager_func)
#def get_sigsapi_choices(self):
# return [ (getattr(object, self.target_class_fk), object) for object in self.target_class.objects.all() ]
def formfield(self, *args, **kwargs):
"""define o FormField a ser usado para esse Field
kwargs contém as opções fornecidas pelo usuário para esse FormField
Returns:
django.forms.Field: o forms.Field apropriado para esse db.Field
"""
defaults = {
'form_class': ModelChoiceField,
#'choices_form_class': ModelChoiceField,
'queryset': self.target_class.objects.all(),
'to_field_name': self.target_class_fk,
#'choices': lambda _self: [ (getattr(object, self.target_class_fk), object) for object in _self.target_class.objects.all() ]
**kwargs,
'widget': Select,
'blank': False,
}
return Field.formfield(self, **defaults)
from django.utils.functional import cached_property
from .query import SIGsAPIModelManager, SIGsAPIRelationModelManager
import inspect
MULTIPLE_OBJECTS_RETURNED = "MultipleObjectsReturned"
DOES_NOT_EXIST = "DoesNotExist"
class ObjectDoesNotExist(Exception):
"""Objeto não encontrad"""
pass
class MultipleObjectsReturned(Exception):
"""Mais de um objeto foi encontrado onde só um era esperado"""
pass
class SIGsAPIOptions:
""" Essa classe será usada como atributo _meta dos SIGsAPIModels
"""
def __init__(self, model, servico = None, recurso = None, path = None, api_query_fields_list = [], pk_fields = []):
self.model = model
self.servico = servico
self.recurso = recurso
self.path = path
self.fields = {}
self.name = model.__class__.__name__
self.api_query_fields_list = api_query_fields_list
self.pk_fields = pk_fields or []
def set_attribute(self, attribute_name, attribute):
self.fields[attribute_name] = attribute
def get_attribute(self, attribute_name):
return self.fields[attribute_name]
def get_field(self, field_name):
return self.get_attribute(field_name)
def _has_contribute_to_class(value):
# Only call contribute_to_class() if it's bound.
return not inspect.isclass(value) and hasattr(value, "contribute_to_sigs_api_class")
class SIGsAPIModelBase(type):
""" Metaclass dos SIGsModel
Essa metaclasse foi criada para realizar algumas operações no momento da criação das classes que herdam
de SIGsModel. Basicamente, os objetivos são:
a) Inserir o Manager (.objects)
b) Transformar os atributos do tipo SIGsAPIField em propriedades calculadas
Args:
type (_type_): _description_
"""
def __new__(cls, name, bases, attrs, **kwargs):
""" Método que cria uma nova classe dessa metaclasse
Por exemplo, se uma classe for definida assim:
```
class TheClass(MyParentClass, metaclass = SIGsAPIModelBase)
my_attr = 32
def a_method(self):
return my_attr + 5
```
No momento em que essa classe estiver sendo criada, esse método será executado
e poderá alterar o nome, as bases, os atributos e métodos dessa classe. Assim,
antes de criarmos a classe, vamos alterar seus atributos para alcançarmos os
dois objetivos acima definidos.
Código fortemente inspirado em django.db.models.base.ModelBase !
Args:
name (_type_): Nome da nova classe (no exemplo, "TheClass")
bases (_type_): Classes das quais ela herda (no exemplo, [MyParentClass,])
attrs (_type_): Os atributos da classe, no exemplo, [my_attr, a_method]
Returns:
_type_: A nova classe!
"""
# Salva o metodo new da metaclasse default do pyton, para chamálo depois
super_new = super().__new__
# esse código limita a execução desse código para classes onde ao menos um dos
# parents é uma instância de SIGsAPIModelBase!
parents = [b for b in bases if isinstance(b, SIGsAPIModelBase)]
if not parents:
return super_new(cls, name, bases, attrs)
# Removemos alguns atributos especiais da classe e colocamos no array new_attrs
# Os atributos especiais são ___module___ e __classcell__
module = attrs.pop("__module__")
new_attrs = {"__module__": module} # esse dicionário guarda os novos atributos
classcell = attrs.pop("__classcell__", None)
if classcell is not None:
new_attrs["__classcell__"] = classcell
# Retiramos, se houver, o atributo Meta, porque vamos precisar modificá-lo
attr_meta = attrs.pop("Meta", None)
# Agora vamos separar os demais atributos. Aqueles que tiverem a propriedade is_sigs_api_field
# serão substituídos, os demais incluídos.
fields_attrs = {} # Esse dicionário guarda os atributos especiais
for obj_name, obj in attrs.items():
if _has_contribute_to_class(obj): # esse método verifica se é um atributo especial
fields_attrs[obj_name] = obj
else:
new_attrs[obj_name] = obj
# Aqui chamamos o método new da metaclasse default do python
new_class = super_new(cls, name, bases, new_attrs)
# ESSE AQUI FOI O CÓDIGO QUE EU NÃO COPIEI DO DJANGO
# Aqui vamos realizar o segundo objetivo
options = {
'model': new_class,
'servico': attr_meta.servico,
'recurso': attr_meta.recurso,
}
if hasattr(attr_meta, "path"):
options['path'] = attr_meta.path
if hasattr(attr_meta, "pk_fields"):
options['pk_fields'] = attr_meta.pk_fields
if hasattr(attr_meta, "api_query_fields_list"):
options['api_query_fields_list'] = attr_meta.api_query_fields_list
_meta = SIGsAPIOptions(**options )
setattr(new_class, '_meta', _meta)
# Aqui vamos realizar o primeiro objetivo, que adicionar o objeto "objects" à
objects = SIGsAPIModelManager(the_class = new_class)
setattr(new_class, 'objects', objects)
# ADicionando as classes de erro DoesNotExist e MultipleObjectsReturned
setattr(new_class, DOES_NOT_EXIST, type(
DOES_NOT_EXIST,
(
ObjectDoesNotExist,
),
{
'__module__': module,
"__qualname__": "{base}.{additive}".format(base = new_class.__qualname__, additive = DOES_NOT_EXIST),
}
))
setattr(new_class, MULTIPLE_OBJECTS_RETURNED, type(
MULTIPLE_OBJECTS_RETURNED,
(
MultipleObjectsReturned,
),
{
'__module__': module,
"__qualname__": "{base}.{additive}".format(base = new_class.__qualname__, additive = MULTIPLE_OBJECTS_RETURNED),
}
))
# Aqui adicionamos os atributos especiais à classe
# to the class.
for field_name, field in fields_attrs.items():
field.contribute_to_sigs_api_class(new_class, field_name)
# Por fim, retornamos a nova classe criada.
return new_class
class SIGsAPIModel(metaclass = SIGsAPIModelBase):
def __init__(self, object, *args, **kwargs):
self._object = object
def serializable_value(self, field_name):
"""
Return the value of the field name for this instance. If the field is
a foreign key, return the id value instead of the object. If there's
no Field object with this name on the model, return the model
attribute's value.
Used to serialize a field's value (in the serializer, or form output,
for example). Normally, you would just access the attribute directly
and not use this method.
"""
try:
field = self._meta.get_field(field_name)
except:
return getattr(self, field_name)
return getattr(self, field.property_name)
class SIGsAPIModelDetalhe(metaclass = SIGsAPIModelBase):
def __init__(self, object, *args, **kwargs):
self._object = object
import requests
from django.conf import settings
from collections import Sequence
class AuthenticationException (Exception):
pass
class APIConnection:
def __init__(self,
servico,
recurso,
path = ""):
self.token = None
self.servico = servico
self.recurso = recurso
self.path = path
def _do_get(self, pk, query, headers = {}):
if pk:
path = str(pk) + "/" + self.path
url = settings.SIGS_API_BASE_URL.format(servico = self.servico, recurso = self.recurso, path = path)
return requests.get(
url,
headers= {
'Authorization': "Bearer {}".format(self.token),
'x-api-key': settings.SIGS_API_KEY,
**headers,
}
)
else:
path = self.path
url = settings.SIGS_API_BASE_URL.format(servico = self.servico, recurso = self.recurso, path = path)
return requests.get(
url,
params = query,
headers= {
'Authorization': "Bearer {}".format(self.token),
'x-api-key': settings.SIGS_API_KEY,
**headers,
}
)
def _authenticate(self):
form_data = {
'grant_type': 'client_credentials',
'client_id': settings.SIGS_API_CLIENT_ID,
'client_secret': settings.SIGS_API_CLIENT_SECRET,
}
headers = {
'x-api-key': settings.SIGS_API_KEY,
'Content-Type': 'application/x-www-form-urlencoded',
}
response = requests.post(
settings.SIGS_API_AUTHENTICATION_URL,
data = form_data,
headers = headers,
)
self.token = response.json()['access_token']
def get(self, pk = None, query = {}, headers = {}):
if not self.token:
self._authenticate()
try:
r = self._do_get(pk, query, headers)
if r.status_code in [401]:
raise AuthenticationException()
return r
except:
try:
self._authenticate()
return self._do_get(pk, query, headers)
except:
raise ConnectionRefusedError()
class BasicFilter:
def __init__(self, key, value = None):
self.keys = key.split('__')
self.value = value
def __call__(self, object) -> bool:
for key in self.keys:
object = getattr(object, key)
if self.value == None:
return key in object
else:
return object == self.value
def __str__(self) -> str:
return '{keys} == {value}'.format(keys = self.keys, value = self.value)
class SIGsAPIQuerySet:
def __init__(self,
the_class,
api_filters = {},
basic_filters = [],
limit = 100,
_results = [],
_last_response = None,
_last_results = [],
api_query_fields_list = [],
):
self.the_class = the_class
if not self.the_class._meta.path:
self.connection = APIConnection(
servico = the_class._meta.servico,
recurso = the_class._meta.recurso,
)
else:
self.connection = APIConnection(
servico = the_class._meta.servico,
recurso = the_class._meta.recurso,
path = the_class._meta.path,
)
self.api_filters = api_filters
self.basic_filters = basic_filters
self.limit = limit
self._results = _results
self._last_response = _last_response
self._last_results = _last_results
self.api_query_fields_list = api_query_fields_list or self.the_class._meta.api_query_fields_list or []
self.stop_iteration_flag = False
self.offset = 0
self._prefetch_related_lookups = False
self.pk = None
def copy(self):
return SIGsAPIQuerySet(
the_class=self.the_class,
api_filters={**self.api_filters},
basic_filters=[ x for x in self.basic_filters],
limit = self.limit,
_results = [ x for x in self._results ],
_last_response = self._last_response,
_last_results = [ x for x in self._last_results],
api_query_fields_list= [x for x in self.api_query_fields_list]
)
def all(self):
return self # Faça nada...
def get(self, pk = None, **kwargs):
if pk: #Execute imediatamente
self.pk = pk
if self.pk:
response = self.connection.get(pk = self.pk, query = {})
object = response.json()
try:
return self.the_class(object[0])
except KeyError:
return self.the_class(object)
elif kwargs: # realize um filtro e pegue o único resultado
return self.filter(**kwargs).get()
else:
self._restart()
self._fetch_all()
if len(self._results)>1:
raise self.the_class.MultipleObjectsReturned()
elif len(self._results)<1:
raise self.the_class.DoesNotExist()
else:
return self._results[0]
def filter(self, **kwargs):
new = self.copy()
for key, value in kwargs.items():
if key in self.the_class._meta.pk_fields:
new.pk = value
elif len(self.api_query_fields_list) == 0 or key in self.api_query_fields_list:
key = key.replace('_','-')
new.api_filters[key] = value
else:
new.basic_filters.append(BasicFilter(key, value))
return new
def _restart(self):
self.offset = 0
self._results = []
def _single_fetch(self):
response = self.connection.get(query = {
'offset': self.offset,
'limit': self.limit,
**self.api_filters,
})
self._last_response = response
def _save_results(self):
if not self._last_response:
return
objects = self._last_response.json()
#if len(objects) == 0:
# self.stop_iteration_flag = True
objects = [ self.the_class(o) for o in objects ]
objects = [ o for o in objects if self._filter(o)]
self._last_results = objects
self._results = self._results + objects
# self._last_response = None
def _next(self):
self._single_fetch()
self.offset += self.limit
if len(self._last_response.json()) == 0:
return False
self._save_results()
return True
def _filter(self, o):
return all( [ condition(o) for condition in self.basic_filters ])
def count(self):
return self.__len__()
def _len_api(self):
response = self.connection.get(query = {
'offset': self.offset,
'limit': self.limit,
**self.api_filters,
}, headers = {
'paginado': 'true',
})
return int(float(response.headers['X-Total']))
def _getitem_api(self, i):
response = self.connection.get(query = {
'offset': i ,
'limit': 1,
**self.api_filters,
})
return response.json()[0]
# Métodos especiais do Python
def __getitem__(self, i):
if isinstance(i, slice):
self._fetch_all()
return self._results[i]
elif i < len(self._results):
return self._results[i]
elif ((i - len(self._results)) > 100) and ( len(self.basic_filters) == 0):
return self.the_class(self._getitem_api(i))
else:
while i >= len(self._results):
self._next()
return self._results[i]
def __len__(self):
if len(self.basic_filters) == 0:
return self._len_api()
else:
self.fetch_all()
return len(self._results)
def __iter__(self):
self._restart()
while self._next():
for r in self._last_results:
yield r
def _fetch_all(self):
self._restart()
while self._next():
pass
def iterator(self):
self._fetch_all()
for r in self._results:
yield r
@property
def model(self):
return self.the_class
class SIGsAPIModelManager:
def __init__(self, the_class):
self.queryset = SIGsAPIQuerySet(the_class=the_class)
def all(self):
return self.queryset.all()
def get(self, *args, **kwargs):
return self.queryset.get(*args, **kwargs)
def filter(self, *args, **kwargs):
return self.queryset.filter(*args, **kwargs)
class SIGsAPIRelationModelManager(SIGsAPIModelManager):
def __init__(self,
the_class,
filters):
super().__init__(the_class = the_class)
self.queryset = self.queryset.filter(**filters)
\ No newline at end of file
class Registry(object):
def __init__(self):
self._registry = {}
self._field_registry = {}
def register(self, cls):
from .schema import SIGsObjectType
assert issubclass(
cls, SIGsObjectType
), 'Apenas SIGsObjectTypes podem ser registrados, received "{}"'.format(
cls.__name__
)
self._registry[cls._meta.model] = cls
def get_type_for_model(self, model):
return self._registry.get(model)
def register_converted_field(self, field, converted):
self._field_registry[field] = converted
def get_converted_field(self, field):
return self._field_registry.get(field)
registry = None
def get_global_registry():
global registry
if not registry:
registry = Registry()
return registry
def reset_global_registry():
global registry
registry = None
from graphene_django.converter import convert_django_field
from .registry import get_global_registry
from .models import SIGsAPIModel
from .fields import SIGsAPIForeignKey, SIGsAPIField, SIGsAPIStringField, SIGsAPIDateTimeField, SIGsAPIDetailsField, SIGsAPIIdField, SIGsAPIManyToOne, SIGsAPIOneToOneField, SIGSAPIOneToManyField, SIGsToOneRelation, SIGsToManyRelation
from graphene import String, Field, Dynamic, List
from graphene.types.objecttype import ObjectType, ObjectTypeOptions
import inspect
from functools import singledispatch
from collections import OrderedDict
from graphene.types.utils import yank_fields_from_attrs
# Conversão dos Fields para Django Models
@convert_django_field.register(SIGsAPIForeignKey)
def convert_field_to_sigs_model(field, registry=None):
model = field.target_class
# Ignore o registry do graphene_django e use o registry do SIGsAPI
registry = get_global_registry()
def dynamic_type():
_type = registry.get_type_for_model(model)
if not _type:
return
return Field(_type, required = not field.null)
return Dynamic(dynamic_type)
# Conversão dos Fields para SIGs Models
@singledispatch
def convert_sigs_field(field, registry):
raise Exception("Tipo do field {field} desconhecido, não é possível converter para Graphene Field.".format(field = field))
@convert_sigs_field.register(SIGsAPIStringField)
@convert_sigs_field.register(SIGsAPIIdField)
def convert_sigs_field_to_string(field, registry):
return String()
@convert_sigs_field.register(SIGsToOneRelation)
@convert_sigs_field.register(SIGsAPIManyToOne)
def convert_sigs_field_to_sigsmodel(field, registry):
model = field.target_class
def dynamic_type():
_type = registry.get_type_for_model(model)
if not _type:
return
return Field(_type, required=not field.null)
return Dynamic(dynamic_type)
@convert_sigs_field.register(SIGsToManyRelation)
@convert_sigs_field.register(SIGSAPIOneToManyField)
def convert_sigs_field_to_sigsmodel(field, registry):
model = field.target_class
def dynamic_type():
_type = registry.get_type_for_model(model)
if not _type:
return
return List(_type, required=not field.null)
return Dynamic(dynamic_type)
def construct_fields(
model,
registry
):
_model_fields = model._meta.fields
fields = OrderedDict()
for name, field in _model_fields.items():
fields[name] = convert_sigs_field(field, registry)
return fields
class SIGsObjectTypeOptions(ObjectTypeOptions):
model = None
# SIGsObjectType
class SIGsObjectType(ObjectType):
@classmethod
def __init_subclass_with_meta__(
cls,
model=None,
fields=None,
interfaces=(),
**options,
):
registry = get_global_registry()
sigs_fields = yank_fields_from_attrs(
construct_fields(model, registry),
_as=Field,
)
_meta = SIGsObjectTypeOptions(cls)
_meta.model = model
_meta.fields = sigs_fields
super(SIGsObjectType, cls).__init_subclass_with_meta__(
_meta=_meta, interfaces=interfaces, **options
)
registry.register(cls)
def resolve_id(self, info):
return self.pk
@classmethod
def is_type_of(cls, root, info):
if isinstance(root, cls):
return True
if not inspect.isclass(root.__class__) and issubclass(root.__class__, SIGsAPIModel):
raise Exception(('Instância não pertence a uma classe SIGsModel "{}".').format(root))
model = root._meta.model #._meta.concrete_model
return model == cls._meta.model
@classmethod
def get_queryset(cls, queryset, info):
return queryset
@classmethod
def get_node(cls, info, id):
queryset = cls.get_queryset(cls._meta.model.objects, info)
try:
return queryset.get(pk=id)
except cls._meta.model.DoesNotExist:
return None
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment