Source code for pysqllike.generic.column_type

# -*- coding: utf-8 -*-
"""
Classes which defines column for class :class:`IterRow <pysqllike.generic.iter_rows.IterRow>`


:githublink:`%|py|6`
"""
from inspect import isfunction
from .iter_exceptions import IterException, NotAllowedOperation
from .others_types import long, NA, EmptyGroup, GroupByContainer

from .column_operator import OperatorId, OperatorMul, ColumnOperator, OperatorAdd
from .column_operator import OperatorDiv, OperatorPow, OperatorSub, OperatorDivN, OperatorMod
from .column_operator import OperatorEq, OperatorNe, OperatorGe, OperatorLe, OperatorGt, OperatorLt
from .column_operator import OperatorNot, OperatorOr, OperatorAnd
from .column_operator import OperatorFunc

from .column_group_operator import OperatorGroupLen, OperatorGroupAvg


[docs]def private_function_type(): "no documentation" pass
[docs]class ColumnType: """ Defines a column of a table. :githublink:`%|py|28` """ _default_name = "__unk__" _str_type = {int: 'int', long: 'long', NA: 'NA', float: 'float', str: 'str', type(private_function_type): 'func', }
[docs] def IsColumnType(self): """ checks it is a column type which used by an operator :githublink:`%|py|39` """ return True
@property def ShortName(self): """ a short name (tells the column type) :githublink:`%|py|46` """ return "any" @property def Name(self): """ property :githublink:`%|py|53` """ return self._name @property def Type(self): """ property :githublink:`%|py|60` """ return self._type @property def Parent(self): """ property :githublink:`%|py|67` """ return self._parent @property def Func(self): """ property :githublink:`%|py|74` """ return self._func
[docs] def __init__( self, name, typ, func=None, parent=tuple(), op=None, owner=None): """ initiates the column :param name: name of the column :param typ: type of the data it will contain (can be None) :param func: a function, if None, if will be the identity :param parent: precise a list of parents if this column was part of a formula :param op: operator to apply between the column :param owner: table which contains the column (only for further validation) function is a function: ``f: x --> y`` :githublink:`%|py|90` """ self._name = name self._type = typ self._value = None self._parent = parent self._op = op self._owner = owner if not isinstance(op, ColumnOperator): raise IterException( "op should be a ColumnOperator not: {0}".format( type(op))) if not isinstance(parent, tuple): raise TypeError("we expect a tuple for parameter parent") for p in parent: p.IsColumnType() if typ not in [int, float, long, str, None, NA, type(private_function_type)]: raise IterException( "type should in [int,float,str,long,function]: " + str(typ)) if isfunction(func): self._func = func elif func is None: self._func = None else: raise IterException( "type of func should in [int,float,str,long,function]: {0}".format( str(func))) if "_func" not in self.__dict__: raise IterException("this column is missing a function")
[docs] def __str__(self): """ usual :githublink:`%|py|129` """ ps = "|".join([_.ShortName for _ in self._parent]) if self._value is not None: return "CT({0},<{1}>,op:{2},#P={3})={4}".format( self._name, ColumnType._str_type[self._type], str(self._op), ps, self()) else: return "CT({0},<{1}>,op:{2},#P={3}) [no loop started]".format( self._name, ColumnType._str_type[self._type], str(self._op), ps)
[docs] def __call__(self): """ returns func(value) :githublink:`%|py|141` """ if self._func is None: if len(self._parent) == 0: if self._value is None: raise ValueError( "method set must be called before for column {0}".format( str(self))) else: res = self._value elif self._op is None: raise ValueError( "there are parents but no operator for column {0}\nParents:\n{1}".format( str(self), self.print_parent())) else: try: res = self._op(self._parent) except TypeError as e: raise IterException( "unable(1) to apply an operator for column op=<{0}>, col={1}, TYPE={2} TYPE_OP={3} TYPE_PARENT={4}".format( str( self._op), str(self), type(self), type( self._op), type( self._parent))) from e except AttributeError as ee: raise IterException( "unable(2) to apply an operator for column op=<{0}>, col={1}, TYPE={2} TYPE_OP={3} TYPE_PARENT={4}".format( str( self._op), str(self), type(self), type( self._op), type( self._parent))) from ee if isinstance(res, ColumnType): raise IterException( "this evaluation(*) cannot return a ColumnType for this column: {0}".format(str(self))) else: # we use a shortcut try: res = self._func(self._value) except TypeError as e: raise IterException( "unable to compute the value of {0}\n{1}".format( str(self), self.print_parent())) from e if isinstance(res, ColumnType): raise IterException( "this evaluation cannot return a ColumnType for this column: {0}".format( str(self))) self.set(res) return res
[docs] def set(self, value): """ Sets a value for this column. :param value: anything in ``[int, float, long, str, function]`` :githublink:`%|py|199` """ if isinstance(value, (int, str, float, long, NA)): self._value = value elif isinstance(value, EmptyGroup): # for an empty group self._value = value elif isinstance(value, list): # for a group self._value = value else: raise IterException( "type of value should be in [int,float,str,long] not {0} for the column {1}".format( type(value), str(self)))
[docs] def set_none(self): """ After a loop on a database, we should put None back as a value. :githublink:`%|py|217` """ for p in self._parent: p.set_none() self._value = None
[docs] def set_name(self, new_name): """ Changes the name of the column. :param newname: new name :githublink:`%|py|227` """ self._name = new_name
[docs] def set_owner(self, new_owner): """ Changes the owner of the column. :param newname: new name :githublink:`%|py|235` """ self._owner = new_owner
[docs] def print_parent(self): """ Returns a string showing the dependencies of this columns. Example:: this_columns parent1 parent11 parent12 parent2 :githublink:`%|py|250` """ if self._parent is None: return self.__str__() else: rows = [self.__str__()] for p in self._parent: rs = [" " + _ for _ in p.print_parent().split("\n")] rows.extend(rs) return "\n".join(rows)
###################################### # functions which create other columns ######################################
[docs] def copy(self, new_owner): """ Returns a copy of this class. :param new_owner: new owner :return: ColumnType :githublink:`%|py|270` """ return ColumnType(self._name, self._type, func=None, parent=( self,), op=OperatorId(), owner=new_owner)
####################################### # operations #######################################
[docs] def __mul__(self, column): """ These operators should be able to translate an expression into function operating on the values. :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|285` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorMul()) else: return self.__mul__(ColumnConstantType(column))
[docs] def __add__(self, column): """ These operators should be able to translate an expression into function operating on the values. :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|299` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorAdd()) else: return self.__add__(ColumnConstantType(column))
[docs] def __sub__(self, column): """ These operators should be able to translate an expression into function operating on the values. :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|313` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorSub()) else: return self.__sub__(ColumnConstantType(column))
[docs] def __truediv__(self, column): """ These operators should be able to translate an expression into function operating on the values. :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|327` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorDiv()) else: return self.__truediv__(ColumnConstantType(column))
[docs] def __floordiv__(self, column): """ These operators should be able to translate an expression into function operating on the values. :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|341` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorDivN()) else: return self.__floordiv__(ColumnConstantType(column))
[docs] def __mod__(self, column): """ these operators should be able to translate an expression into function operating on the values :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|355` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorMod()) else: return self.__mod__(ColumnConstantType(column))
[docs] def __pow__(self, column): """ these operators should be able to translate an expression into function operating on the values :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|369` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorPow()) else: return self.__pow__(ColumnConstantType(column))
####################################### # test #######################################
[docs] def __eq__(self, column): """ these operators should be able to translate an expression into function operating on the values :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|387` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorEq()) else: return self.__eq__(ColumnConstantType(column))
[docs] def __lt__(self, column): """ these operators should be able to translate an expression into function operating on the values :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|401` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorLt()) else: return self.__lt__(ColumnConstantType(column))
[docs] def __le__(self, column): """ these operators should be able to translate an expression into function operating on the values :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|415` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorLe()) else: return self.__le__(ColumnConstantType(column))
[docs] def __gt__(self, column): """ these operators should be able to translate an expression into function operating on the values :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|429` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorGt()) else: return self.__gt__(ColumnConstantType(column))
[docs] def __ge__(self, column): """ these operators should be able to translate an expression into function operating on the values :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|443` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorGe()) else: return self.__ge__(ColumnConstantType(column))
[docs] def __ne__(self, column): """ these operators should be able to translate an expression into function operating on the values :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|457` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorNe()) else: return self.__ne__(ColumnConstantType(column))
####################################### # logical #######################################
[docs] def Not(self): """ ``not`` cannot be overriden :githublink:`%|py|471` """ return self.__not__()
[docs] def __not__(self): """ these operators should be able to translate an expression into function operating on the values :return: a ColumnType :githublink:`%|py|480` """ return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self,), op=OperatorNot())
[docs] def Or(self, column): """ ``or`` cannot be overriden :githublink:`%|py|487` """ return self.__or__(column)
[docs] def __or__(self, column): """ these operators should be able to translate an expression into function operating on the values :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|497` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorOr()) else: return self.__or__(ColumnConstantType(column))
[docs] def And(self, column): """ ``and`` cannot be overriden :githublink:`%|py|507` """ return self.__and__(column)
[docs] def __and__(self, column): """ these operators should be able to translate an expression into function operating on the values :param column: a function or an int or a float or a long or a str or a ColumnType :return: a ColumnType :githublink:`%|py|517` """ if isinstance(column, ColumnType): return ColumnType(ColumnType._default_name, self._type, func=None, parent=( self, column), op=OperatorAnd()) else: return self.__and__(ColumnConstantType(column))
####################################### # group function #######################################
[docs] def len(self): """ returns a group columns to count the number of observations :githublink:`%|py|531` """ return ColumnGroupType( ColumnType._default_name, int, parent=(self,), op=OperatorGroupLen())
[docs] def count(self): """ returns a group columns to count the number of observations :githublink:`%|py|538` """ return self.len()
[docs] def avg(self): """ returns a group columns to return an average :githublink:`%|py|544` """ return ColumnGroupType( ColumnType._default_name, float, parent=(self,), op=OperatorGroupAvg())
[docs]class ColumnConstantType(ColumnType): """ defines a constant as a column :githublink:`%|py|553` """
[docs] def __init__(self, const): self._value = const self._func = lambda x, c=self._value: c self._parent = None self._op = None self._type = type(const) self._const = const self._owner = None if isinstance(const, (int, float, long, str, NA)): pass else: raise ValueError( "this value is not a constant: {0}".format( str(const)))
@property def ShortName(self): """ a short name (tells the column type) :githublink:`%|py|575` """ return "cst"
[docs] def set_none(self): """ do nothing (it is a constant) :githublink:`%|py|581` """ pass
[docs] def set(self, value): """ do nothing (it is a constant) :param value: anything in [int,float,long,str, function ] :githublink:`%|py|589` """ pass
[docs] def __call__(self): """ return the constant :githublink:`%|py|595` """ return self._const
[docs] def __str__(self): """ usual :githublink:`%|py|601` """ return "cst({0})".format(self())
[docs]class ColumnTableType(ColumnType): """ defines a table column (not coming from an expression) :githublink:`%|py|609` """
[docs] def __init__(self, name, typ, owner): """ constructor :param name: name of the column :param typ: type of the column :param owner: owner of this column :githublink:`%|py|618` """ self._name = name self._func = None self._parent = None self._op = None self._type = typ self._owner = owner
@property def ShortName(self): """ a short name (tells the column type) :githublink:`%|py|630` """ return "col"
[docs] def set_none(self): """ after a loop on a database, we should put None back as a value :githublink:`%|py|636` """ self._value = None
[docs] def __call__(self): """ returns the content :githublink:`%|py|642` """ if self._value is None: raise IterException( "this column should contain a value: {0}".format( str(self))) return self._value
[docs] def __str__(self): """ usual :githublink:`%|py|652` """ return "col({0},{1})".format( self._name, ColumnType._str_type[self._type])
[docs]class ColumnGroupType(ColumnType): """ defines a column which processes a group of rows (after a groupby) :githublink:`%|py|661` """
[docs] def __init__(self, name, typ, parent, op): """ constructor :param name: name of the column :param typ: type of the column :param owner: owner of this column :param op: operator :githublink:`%|py|671` """ self._name = name self._value = None self._parent = parent self._opgr = op self._op = OperatorId() self._type = typ self._owner = None self._func = None
@property def ShortName(self): """ a short name (tells the column type) :githublink:`%|py|685` """ return "group"
[docs] def set_none(self): """ after a loop on a database, we should put None back as a value :githublink:`%|py|691` """ self._value = None
[docs] def __call__(self): """ returns the content :githublink:`%|py|697` """ if isinstance(self._value, GroupByContainer): try: return self._opgr(self._value) except TypeError as e: raise IterException( "unable(1) to apply an operator for column op=<{0}>, col={1}, TYPE={2} TYPE_OP={3}".format( str( self._op), str(self), type(self), type( self._op))) from e except AttributeError as ee: raise IterException( "unable(2) to apply an operator for column op=<{0}>, col={1}, TYPE={2} TYPE_OP={3}".format( str( self._op), str(self), type(self), type( self._op))) from ee else: return super().__call__()
[docs] def __str__(self): """ usual :githublink:`%|py|719` """ return "CGT[{0}]({1})".format(str(self._opgr), self._name)
[docs] def set(self, value): """ sets a value for this column :param value: anything in [int,float,long,str, function ] :githublink:`%|py|727` """ self._value = value if hasattr(value, "__iter__") and \ not isinstance(value, str) and \ not isinstance(value, GroupByContainer): raise IterException( "type of value should be GroupByContainer not {0} for the column {1}".format( type(value), str(self)))
[docs] def __mul__(self, column): """ forbidden :githublink:`%|py|740` """ raise NotAllowedOperation()
[docs] def __add__(self, column): """ forbidden :githublink:`%|py|746` """ raise NotAllowedOperation()
[docs] def __sub__(self, column): """ forbidden :githublink:`%|py|752` """ raise NotAllowedOperation()
[docs] def __truediv__(self, column): """ forbidden :githublink:`%|py|758` """ raise NotAllowedOperation()
[docs] def __floordiv__(self, column): """ forbidden :githublink:`%|py|764` """ raise NotAllowedOperation()
[docs] def __mod__(self, column): """ forbidden :githublink:`%|py|770` """ raise NotAllowedOperation()
[docs] def __pow__(self, column): """ forbidden :githublink:`%|py|776` """ raise NotAllowedOperation()
[docs]class CFT(ColumnType): """ defines a function :githublink:`%|py|784` """
[docs] def __init__(self, func, *l): """ constructor (a function cannot accept keywords) :param func: contained function :param l: list of ColumnType :githublink:`%|py|792` """ self._name = None self._func = None self._parent = None self._op = OperatorFunc(func) self._type = type(private_function_type) self._owner = None self._thisfunc = func self._parent = tuple(l) for _ in l: if not isinstance(_, ColumnType): raise TypeError("expecting a column type, not " + str(type(_)))
@property def ShortName(self): """ a short name (tells the column type) :githublink:`%|py|810` """ return "func"
[docs] def set_none(self): """ after a loop on a database, we should put None back as a value :githublink:`%|py|816` """ self._value = None
[docs] def __str__(self): """ usual :githublink:`%|py|822` """ return "func({0},{1})".format( self._name, ColumnType._str_type[self._type])