# ============================================================================
#
# Copyright (C) 2007-2016 Conceptive Engineering bvba.
# www.conceptive.be / info@conceptive.be
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# * Neither the name of Conceptive Engineering nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
# DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# ============================================================================
"""Set of classes to store persons, organizations, relationships and
contact mechanisms
These structures are modeled like described in 'The Data Model Resource Book'
by Len Silverston, Chapter 2
"""
import datetime
import six
from sqlalchemy.ext import hybrid
from sqlalchemy.types import Date, Unicode
from sqlalchemy.sql.expression import and_
from sqlalchemy import orm, schema, sql, ForeignKey
from camelot.admin.entity_admin import EntityAdmin
from camelot.core.orm import ( Entity, using_options, Field, ManyToMany,
ManyToOne, OneToMany, ColumnProperty )
from camelot.core.utils import ugettext_lazy as _
import camelot.types
from camelot.view.controls import delegates
from camelot.view.forms import Form, TabForm, HBoxForm, WidgetOnlyForm, Stretch
from .authentication import end_of_times
[docs]class GeographicBoundary( Entity ):
"""The base class for Country and City"""
using_options( tablename = 'geographic_boundary' )
code = schema.Column( Unicode( 10 ) )
name = schema.Column( Unicode( 40 ), nullable = False )
row_type = schema.Column( Unicode(40), nullable = False )
__mapper_args__ = { 'polymorphic_on' : row_type }
@ColumnProperty
def full_name( self ):
return self.code + ' ' + self.name
def __unicode__( self ):
return u'%s %s' % ( self.code, self.name )
[docs]class Country( GeographicBoundary ):
"""A subclass of GeographicBoundary used to store the name and the
ISO code of a country"""
using_options( tablename = 'geographic_boundary_country' )
geographicboundary_id = Field( camelot.types.PrimaryKey(),
ForeignKey('geographic_boundary.id'),
primary_key = True )
__mapper_args__ = {'polymorphic_identity': 'country'}
@classmethod
def get_or_create( cls, code, name ):
country = Country.query.filter_by( code = code ).first()
if not country:
country = Country( code = code, name = name )
orm.object_session( country ).flush()
return country
class Admin( EntityAdmin ):
form_size = ( 700, 150 )
verbose_name = _('Country')
verbose_name_plural = _('Countries')
list_display = ['name', 'code']
[docs]class City( GeographicBoundary ):
"""A subclass of GeographicBoundary used to store the name, the postal code
and the Country of a city"""
using_options( tablename = 'geographic_boundary_city' )
country = ManyToOne( Country, required = True, ondelete = 'cascade', onupdate = 'cascade' )
geographicboundary_id = Field( camelot.types.PrimaryKey(),
ForeignKey('geographic_boundary.id'),
primary_key = True )
__mapper_args__ = {'polymorphic_identity': 'city'}
def __unicode__( self ):
if None not in (self.code, self.name, self.country):
return u'{0.code} {0.name} [{1.code}]'.format( self, self.country )
return u''
@classmethod
def get_or_create( cls, country, code, name ):
city = City.query.filter_by( code = code, country = country ).first()
if not city:
city = City( code = code, name = name, country = country )
orm.object_session( city ).flush()
return city
class Admin( EntityAdmin ):
verbose_name = _('City')
verbose_name_plural = _('Cities')
form_size = ( 700, 150 )
list_display = ['code', 'name', 'country']
[docs]class Address( Entity ):
"""The Address to be given to a Party (a Person or an Organization)"""
using_options( tablename = 'address' )
street1 = schema.Column( Unicode( 128 ), nullable = False )
street2 = schema.Column( Unicode( 128 ) )
city = ManyToOne( City,
required = True,
ondelete = 'cascade',
onupdate = 'cascade',
lazy = 'subquery' )
def name( self ):
return sql.select( [self.street1 + ', ' + GeographicBoundary.full_name],
whereclause = (GeographicBoundary.id == self.city_geographicboundary_id))
name = ColumnProperty( name, deferred = True )
@classmethod
def get_or_create( cls, street1, street2, city ):
address = cls.query.filter_by( street1 = street1, street2 = street2, city = city ).first()
if not address:
address = cls( street1 = street1, street2 = street2, city = city )
orm.object_session( address ).flush()
return address
def __unicode__( self ):
return u'%s, %s' % ( self.street1 or '', self.city or '' )
class Admin( EntityAdmin ):
verbose_name = _('Address')
verbose_name_plural = _('Addresses')
list_display = ['street1', 'street2', 'city']
form_size = ( 700, 150 )
field_attributes = {'street1':{'minimal_column_width':30}}
def get_depending_objects( self, address ):
for party_address in address.party_addresses:
yield party_address
if party_address.party != None:
yield party_address.party
class PartyContactMechanismAdmin( EntityAdmin ):
form_size = ( 700, 200 )
verbose_name = _('Contact mechanism')
verbose_name_plural = _('Contact mechanisms')
list_search = ['party_name', 'mechanism']
list_display = ['party_name', 'mechanism', 'comment', 'from_date', ]
form_display = Form( ['mechanism', 'comment', 'from_date', 'thru_date', ] )
field_attributes = {'party_name':{'minimal_column_width':25, 'editable':False},
'comment': {'name': _('Comment')},
'mechanism':{'minimal_column_width':25,
'editable':True,
'nullable':False,
'name':_('Mechanism'),
'delegate':delegates.VirtualAddressDelegate}}
def get_depending_objects(self, contact_mechanism ):
party = contact_mechanism.party
if party and (party not in Party.query.session.new):
yield party
def get_compounding_objects( self, contact_mechanism ):
if contact_mechanism.contact_mechanism:
yield contact_mechanism.contact_mechanism
class PartyPartyContactMechanismAdmin( PartyContactMechanismAdmin ):
list_search = ['party_name', 'mechanism']
list_display = ['mechanism', 'comment', 'from_date', ]
class WithAddresses(object):
@hybrid.hybrid_property
def street1( self ):
return self._get_address_field( u'street1' )
@street1.setter
def street1_setter( self, value ):
return self._set_address_field( u'street1', value )
@street1.expression
def street1_expression(cls):
return sql.select([Address.street1],
whereclause=cls.first_address_filter(),
limit=1).as_scalar()
@hybrid.hybrid_property
def street2( self ):
return self._get_address_field( u'street2' )
@street2.setter
def street2_setter( self, value ):
return self._set_address_field( u'street2', value )
@hybrid.hybrid_property
def city( self ):
return self._get_address_field( u'city' )
@city.setter
def city_setter( self, value ):
return self._set_address_field( u'city', value )
@city.expression
def city_expression(cls):
GB = orm.aliased(GeographicBoundary)
return sql.select([GB.code + ' ' + GB.name],
whereclause=sql.and_(
GB.id==Address.city_geographicboundary_id,
cls.first_address_filter()
),
limit=1).as_scalar()
def get_first_address(self):
raise NotImplementedError()
def set_first_address(self):
raise NotImplementedError()
@classmethod
def first_address_filter(cls):
raise NotImplementedError
def _get_address_field( self, name ):
first_address = self.get_first_address()
if first_address is not None:
return getattr( first_address, name )
def _set_address_field( self, name, value ):
address = self.set_first_address()
setattr( address, name, value )
if address.street1==None and address.street2==None and address.city==None:
session = orm.object_session( address )
if address in session.new:
session.expunge( address )
self.addresses.remove( address )
else:
session.delete( address )
[docs]class Party(Entity, WithAddresses):
"""Base class for persons and organizations. Use this base class to refer to either persons or
organisations in building authentication systems, contact management or CRM"""
using_options( tablename = 'party' )
row_type = schema.Column( Unicode(40), nullable = False )
__mapper_args__ = { 'polymorphic_on' : row_type }
@classmethod
def first_address_filter(cls):
return sql.and_(PartyAddress.party_id==cls.id,
PartyAddress.address_id==Address.id)
@property
def name( self ):
return ''
def get_first_address(self):
for party_address in self.addresses:
return party_address
def set_first_address(self):
if not self.addresses:
address = PartyAddress()
self.addresses.append( address )
return self.addresses[0]
def _get_contact_mechanism( self, described_by ):
"""Get a specific type of contact mechanism
"""
for party_contact_mechanism in self.contact_mechanisms:
contact_mechanism = party_contact_mechanism.contact_mechanism
if contact_mechanism != None:
mechanism = contact_mechanism.mechanism
if mechanism != None:
if mechanism[0] == described_by:
return mechanism
def _set_contact_mechanism( self, described_by, value ):
"""Set a specific type of contact mechanism
"""
assert value[0] in camelot.types.VirtualAddress.virtual_address_types
for party_contact_mechanism in self.contact_mechanisms:
contact_mechanism = party_contact_mechanism.contact_mechanism
if contact_mechanism != None:
mechanism = contact_mechanism.mechanism
if mechanism != None:
if mechanism[0] == described_by:
if value and value[1]:
contact_mechanism.mechanism = value
else:
self.contact_mechanisms.remove( party_contact_mechanism )
session = orm.object_session( party_contact_mechanism )
if (session is not None) and party_contact_mechanism.id:
session.delete( party_contact_mechanism )
return
if value and value[1]:
contact_mechanism = ContactMechanism( mechanism = value )
party_contact_mechanism = PartyContactMechanism( contact_mechanism = contact_mechanism )
self.contact_mechanisms.append( party_contact_mechanism )
@hybrid.hybrid_property
def email( self ):
return self._get_contact_mechanism( u'email' )
@email.setter
def email_setter( self, value ):
return self._set_contact_mechanism( u'email', value )
@email.expression
def email_expression( self ):
return orm.aliased( ContactMechanism ).mechanism
@hybrid.hybrid_property
def phone( self ):
return self._get_contact_mechanism( u'phone' )
@phone.setter
def phone_setter( self, value ):
return self._set_contact_mechanism( u'phone', value )
@phone.expression
def phone_expression( self ):
return orm.aliased( ContactMechanism ).mechanism
@hybrid.hybrid_property
def fax( self ):
return self._get_contact_mechanism( u'fax' )
@fax.setter
def fax_setter( self, value ):
return self._set_contact_mechanism( u'fax', value )
@fax.expression
def fax_expression( self ):
return orm.aliased( ContactMechanism ).mechanism
@hybrid.hybrid_property
def mobile( self ):
return self._get_contact_mechanism( u'mobile' )
@mobile.setter
def mobile_setter( self, value ):
return self._set_contact_mechanism( u'mobile', value )
@mobile.expression
def mobile_expression( self ):
return orm.aliased( ContactMechanism ).mechanism
def full_name( self ):
aliased_organisation = sql.alias( Organization.table )
aliased_person = sql.alias( Person.table )
return sql.functions.coalesce( sql.select( [sql.functions.coalesce(aliased_person.c.first_name,'') + ' ' + sql.functions.coalesce(aliased_person.c.last_name, '')],
whereclause = and_( aliased_person.c.party_id == self.id ),
).limit( 1 ).as_scalar(),
sql.select( [aliased_organisation.c.name],
whereclause = and_( aliased_organisation.c.party_id == self.id ),
).limit( 1 ).as_scalar() )
full_name = ColumnProperty( full_name, deferred=True )
[docs]class Organization( Party ):
"""An organization represents any internal or external organization. Organizations can include
businesses and groups of individuals"""
using_options( tablename = 'organization' )
party_id = Field( camelot.types.PrimaryKey(),
ForeignKey('party.id'),
primary_key = True )
__mapper_args__ = {'polymorphic_identity': u'organization'}
name = schema.Column( Unicode( 50 ), nullable = False, index = True )
logo = schema.Column( camelot.types.Image( upload_to = 'organization-logo' ))
tax_id = schema.Column( Unicode( 20 ) )
def __unicode__( self ):
return self.name or ''
@property
def note(self):
session = orm.object_session(self)
if session is not None:
cls = self.__class__
if session.query(cls).filter( sql.and_( cls.name == self.name,
cls.id != self.id ) ).count():
return _('An organization with the same name already exists')
# begin short person definition
[docs]class Person( Party ):
"""Person represents natural persons
"""
using_options( tablename = 'person' )
party_id = Field( camelot.types.PrimaryKey(),
ForeignKey('party.id'),
primary_key = True )
__mapper_args__ = {'polymorphic_identity': u'person'}
first_name = schema.Column( Unicode( 40 ), nullable = False )
last_name = schema.Column( Unicode( 40 ), nullable = False )
# end short person definition
middle_name = schema.Column( Unicode( 40 ) )
personal_title = schema.Column( Unicode( 10 ) )
suffix = schema.Column( Unicode( 3 ) )
sex = schema.Column( Unicode( 1 ), default = u'M' )
birthdate = schema.Column( Date() )
martial_status = schema.Column( Unicode( 1 ) )
social_security_number = schema.Column( Unicode( 12 ) )
passport_number = schema.Column( Unicode( 20 ) )
passport_expiry_date = schema.Column( Date() )
picture = schema.Column( camelot.types.Image( upload_to = 'person-pictures' ))
comment = schema.Column( camelot.types.RichText() )
@property
def note(self):
for person in self.__class__.query.filter_by(first_name=self.first_name, last_name=self.last_name):
if person != self:
return _('A person with the same name already exists')
@property
def name( self ):
# we don't use full name in here, because for new objects, full name will be None, since
# it needs to be fetched from the db first
return ' '.join([name for name in [self.first_name, self.last_name] if name])
def __unicode__( self ):
return self.name or ''
#class PartyRelationship( Entity ):
#using_options( tablename = 'party_relationship' )
#from_date = Field( Date(), default = datetime.date.today, required = True, index = True )
#thru_date = Field( Date(), default = end_of_times, required = True, index = True )
#comment = Field( camelot.types.RichText() )
#row_type = schema.Column( Unicode(40), nullable = False )
#__mapper_args__ = { 'polymorphic_on' : row_type }
#class Admin( EntityAdmin ):
#verbose_name = _('Relationship')
#verbose_name_plural = _('Relationships')
#list_display = ['from_date', 'thru_date']
#class EmployerEmployee( PartyRelationship ):
#"""Relation from employer to employee"""
#using_options( tablename = 'party_relationship_empl' )
#established_from = ManyToOne( Organization, required = True, ondelete = 'cascade', onupdate = 'cascade',
#backref=orm.backref('employees', cascade='all, delete, delete-orphan' ) ) # the employer
#established_to = ManyToOne( Person, required = True, ondelete = 'cascade', onupdate = 'cascade'
# backref=orm.backref('employers', cascade='all, delete, delete-orphan' )) # the employee
#partyrelationship_id = Field( Integer,
#ForeignKey('party_relationship.id'),
#primary_key = True )
#__mapper_args__ = {'polymorphic_identity': 'employeremployee'}
#@ColumnProperty
#def first_name( self ):
#return sql.select( [Person.first_name], Person.party_id == self.established_to_party_id )
#@ColumnProperty
#def last_name( self ):
#return sql.select( [Person.last_name], Person.party_id == self.established_to_party_id )
#@ColumnProperty
#def social_security_number( self ):
#return sql.select( [Person.social_security_number], Person.party_id == self.established_to_party_id )
#def __unicode__( self ):
#return u'%s %s %s' % ( unicode( self.established_to ), _('Employed by'),unicode( self.established_from ) )
#class Admin( PartyRelationship.Admin ):
#verbose_name = _('Employment relation')
#verbose_name_plural = _('Employment relations')
#list_filter = ['established_from.name']
#list_search = ['established_from.name', 'established_to.first_name', 'established_to.last_name']
#class EmployeeAdmin( EntityAdmin ):
#verbose_name = _('Employee')
#list_display = ['established_to', 'from_date', 'thru_date']
#form_display = ['established_to', 'comment', 'from_date', 'thru_date']
#field_attributes = {'established_to':{'name':_( 'Name' )}}
#class EmployerAdmin( EntityAdmin ):
#verbose_name = _('Employer')
#list_display = ['established_from', 'from_date', 'thru_date']
#form_display = ['established_from', 'comment', 'from_date', 'thru_date']
#field_attributes = {'established_from':{'name':_( 'Name' )}}
#class DirectedDirector( PartyRelationship ):
#"""Relation from a directed organization to a director"""
#using_options( tablename = 'party_relationship_dir' )
#established_from = ManyToOne( Organization, required = True, ondelete = 'cascade', onupdate = 'cascade',
#backref=orm.backref('directors', cascade='all, delete, delete-orphan' ))
#established_to = ManyToOne( Party, required = True, ondelete = 'cascade', onupdate = 'cascade',
#backref=orm.backref('directed_organizations', cascade='all, delete, delete-orphan' ))
#title = Field( Unicode( 256 ) )
#represented_by = OneToMany( 'RepresentedRepresentor', inverse = 'established_to' )
#partyrelationship_id = Field( Integer,
#ForeignKey('party_relationship.id'),
#primary_key = True )
#__mapper_args__ = {'polymorphic_identity': 'directeddirector'}
#class Admin( PartyRelationship.Admin ):
#verbose_name = _('Direction structure')
#verbose_name_plural = _('Direction structures')
#list_display = ['established_from', 'established_to', 'title', 'represented_by']
#list_search = ['established_from.full_name', 'established_to.full_name']
#field_attributes = {'established_from':{'name':_('Organization')},
#'established_to':{'name':_('Director')}}
#class DirectorAdmin( Admin ):
#verbose_name = _('Director')
#list_display = ['established_to', 'title', 'from_date', 'thru_date']
#form_display = ['established_to', 'title', 'from_date', 'thru_date', 'represented_by', 'comment']
#class DirectedAdmin( Admin ):
#verbose_name = _('Directed organization')
#list_display = ['established_from', 'title', 'from_date', 'thru_date']
#form_display = ['established_from', 'title', 'from_date', 'thru_date', 'represented_by', 'comment']
#class RepresentedRepresentor( Entity ):
#"""Relation from a representing party to the person representing the party"""
#using_options( tablename = 'party_representor' )
#from_date = Field( Date(), default = datetime.date.today, required = True, index = True )
#thru_date = Field( Date(), default = end_of_times, required = True, index = True )
#comment = Field( camelot.types.RichText() )
#established_from = ManyToOne( Person, required = True, ondelete = 'cascade', onupdate = 'cascade' )
#established_to = ManyToOne( DirectedDirector, required = True, ondelete = 'cascade', onupdate = 'cascade' )
#class Admin( EntityAdmin ):
#verbose_name = _('Represented by')
#list_display = ['established_from', 'from_date', 'thru_date']
#form_display = ['established_from', 'from_date', 'thru_date', 'comment']
#field_attributes = {'established_from':{'name':_( 'Name' )}}
#class SupplierCustomer( PartyRelationship ):
#"""Relation from supplier to customer"""
#using_options( tablename = 'party_relationship_suppl' )
#established_from = ManyToOne( Party, required = True, ondelete = 'cascade', onupdate = 'cascade',
#backref=orm.backref('customers', cascade='all, delete, delete-orphan' ))
#established_to = ManyToOne( Party, required = True, ondelete = 'cascade', onupdate = 'cascade',
#backref=orm.backref('suppliers', cascade='all, delete, delete-orphan' ))
#partyrelationship_id = Field( Integer,
#ForeignKey('party_relationship.id'),
#primary_key = True )
#__mapper_args__ = {'polymorphic_identity': 'suppliercustomer'}
#class Admin( PartyRelationship.Admin ):
#verbose_name = _('Supplier - Customer')
#list_display = ['established_from', 'established_to', 'from_date', 'thru_date']
#class CustomerAdmin( EntityAdmin ):
#verbose_name = _('Customer')
#list_display = ['established_to', ]
#form_display = ['established_to', 'comment', 'from_date', 'thru_date']
#field_attributes = {'established_to':{'name':_( 'Name' )}}
#class SupplierAdmin( EntityAdmin ):
#verbose_name = _('Supplier')
#list_display = ['established_from', ]
#form_display = ['established_from', 'comment', 'from_date', 'thru_date']
#field_attributes = {'established_from':{'name':_( 'Name' )}}
#class SharedShareholder( PartyRelationship ):
#"""Relation from a shared organization to a shareholder"""
#using_options( tablename = 'party_relationship_shares' )
#established_from = ManyToOne( Organization, required = True, ondelete = 'cascade', onupdate = 'cascade',
#backref=orm.backref('shareholders', cascade='all, delete, delete-orphan' ))
#established_to = ManyToOne( Party, required = True, ondelete = 'cascade', onupdate = 'cascade',
#backref=orm.backref('shares', cascade='all, delete, delete-orphan' ) )
#shares = Field( Integer() )
#partyrelationship_id = Field( Integer,
#ForeignKey('party_relationship.id'),
#primary_key = True )
#__mapper_args__ = {'polymorphic_identity': 'sharedshareholder'}
#class Admin( PartyRelationship.Admin ):
#verbose_name = _('Shareholder structure')
#verbose_name_plural = _('Shareholder structures')
#list_display = ['established_from', 'established_to', 'shares',]
#list_search = ['established_from.full_name', 'established_to.full_name']
#field_attributes = {'established_from':{'name':_('Organization')},
#'established_to':{'name':_('Shareholder')}}
#class ShareholderAdmin( Admin ):
#verbose_name = _('Shareholder')
#list_display = ['established_to', 'shares', 'from_date', 'thru_date']
#form_display = ['established_to', 'shares', 'from_date', 'thru_date', 'comment']
#form_size = (500, 300)
#class SharedAdmin( Admin ):
#verbose_name = _('Shares')
#verbose_name_plural = _('Shares')
#list_display = ['established_from', 'shares', 'from_date', 'thru_date']
#form_display = ['established_from', 'shares', 'from_date', 'thru_date', 'comment']
#form_size = (500, 300)
class Addressable(object):
def _get_address_field( self, name ):
if self.address:
return getattr( self.address, name )
def _set_address_field( self, name, value ):
if not self.address:
self.address = Address()
setattr( self.address, name, value )
@hybrid.hybrid_property
def street1( self ):
return self._get_address_field( u'street1' )
@street1.setter
def street1_setter( self, value ):
return self._set_address_field( u'street1', value )
@street1.expression
def street1_expression( self ):
return Address.street1
@hybrid.hybrid_property
def street2( self ):
return self._get_address_field( u'street2' )
@street2.expression
def street2_expression( self ):
return Address.street2
@street2.setter
def street2_setter( self, value ):
return self._set_address_field( u'street2', value )
@hybrid.hybrid_property
def city( self ):
return self._get_address_field( u'city' )
@city.setter
def city_setter( self, value ):
return self._set_address_field( u'city', value )
class Admin(object):
field_attributes = dict(
street1 = dict( editable = True,
name = _('Street'),
minimal_column_width = 50 ),
street2 = dict( editable = True,
name = _('Street Extra'),
minimal_column_width = 50 ),
city = dict( editable = True,
delegate = delegates.Many2OneDelegate,
target = City ),
email = dict( editable = True,
minimal_column_width = 20,
name = _('Email'),
address_type = 'email',
from_string = lambda s:('email', s),
delegate = delegates.VirtualAddressDelegate),
phone = dict( editable = True,
minimal_column_width = 20,
address_type = 'phone',
name = _('Phone'),
from_string = lambda s:('phone', s),
delegate = delegates.VirtualAddressDelegate ),
mobile = dict( editable = True,
minimal_column_width = 20,
address_type = 'mobile',
name = _('Mobile'),
from_string = lambda s:('mobile', s),
delegate = delegates.VirtualAddressDelegate ),
fax = dict( editable = True,
minimal_column_width = 20,
address_type = 'fax',
name = _('Fax'),
from_string = lambda s:('fax', s),
delegate = delegates.VirtualAddressDelegate ), )
class PartyAddress( Entity, Addressable ):
using_options( tablename = 'party_address' )
party = ManyToOne( Party,
required = True,
ondelete = 'cascade',
onupdate = 'cascade',
lazy = 'subquery',
backref = orm.backref('addresses', lazy = True,
cascade='all, delete, delete-orphan'))
address = ManyToOne( Address,
required = True,
backref = 'party_addresses',
ondelete = 'cascade',
onupdate = 'cascade',
lazy = 'subquery' )
from_date = schema.Column( Date(), default = datetime.date.today, nullable=False, index = True )
thru_date = schema.Column( Date(), default = end_of_times, nullable=False, index = True )
comment = schema.Column( Unicode( 256 ) )
def party_name( self ):
return sql.select( [sql.func.coalesce(Party.full_name, '')],
whereclause = (Party.id==self.party_id))
party_name = ColumnProperty( party_name, deferred = True )
def __unicode__( self ):
return '%s : %s' % ( six.text_type( self.party ), six.text_type( self.address ) )
class Admin( EntityAdmin ):
verbose_name = _('Address')
verbose_name_plural = _('Addresses')
list_search = ['party_name', 'street1', 'street2',]
list_display = ['party_name', 'street1', 'street2', 'city', 'comment']
form_display = [ 'party', 'street1', 'street2', 'city', 'comment',
'from_date', 'thru_date']
form_size = ( 700, 200 )
field_attributes = dict(party_name=dict(editable=False, name='Party', minimal_column_width=30))
def get_compounding_objects( self, party_address ):
if party_address.address!=None:
yield party_address.address
[docs]class AddressAdmin( PartyAddress.Admin ):
"""Admin with only the Address information and not the Party information"""
verbose_name = _('Address')
list_display = ['street1', 'city', 'comment']
form_display = ['street1', 'street2', 'city', 'comment', 'from_date', 'thru_date']
field_attributes = dict(street1 = dict(name=_('Street'),
editable=True,
nullable=False),
street2 = dict(name=_('Street Extra'),
editable=True),
city = dict(name=_('City'),
editable=True,
nullable=False,
delegate=delegates.Many2OneDelegate,
target=City),
)
def get_depending_objects( self, party_address ):
if party_address.party:
yield party_address.party
class PartyAddressRoleType( Entity ):
using_options( tablename = 'party_address_role_type' )
code = schema.Column( Unicode( 10 ) )
description = schema.Column( Unicode( 40 ) )
class Admin( EntityAdmin ):
verbose_name = _('Address role type')
list_display = ['code', 'description']
class ContactMechanism( Entity ):
using_options( tablename = 'contact_mechanism' )
mechanism = schema.Column( camelot.types.VirtualAddress( 256 ), nullable = False )
party_address = ManyToOne( PartyAddress, ondelete = 'set null', onupdate = 'cascade' )
party_contact_mechanisms = OneToMany( 'PartyContactMechanism' )
def __unicode__( self ):
if self.mechanism:
return u'%s : %s' % ( self.mechanism[0], self.mechanism[1] )
class Admin( EntityAdmin ):
form_size = ( 700, 150 )
verbose_name = _('Contact mechanism')
list_display = ['mechanism']
form_display = Form( ['mechanism', 'party_address'] )
field_attributes = {'mechanism':{'minimal_column_width':25}}
def get_depending_objects(self, contact_mechanism ):
for party_contact_mechanism in contact_mechanism.party_contact_mechanisms:
yield party_contact_mechanism
party = party_contact_mechanism.party
if party:
yield party
class PartyContactMechanism( Entity ):
using_options( tablename = 'party_contact_mechanism' )
party = ManyToOne( Party, required = True, ondelete = 'cascade', onupdate = 'cascade',
backref = orm.backref('contact_mechanisms', lazy = 'select',
cascade='all, delete, delete-orphan' )
)
contact_mechanism = ManyToOne( ContactMechanism, lazy='joined', required = True, ondelete = 'cascade', onupdate = 'cascade' )
from_date = schema.Column( Date(), default = datetime.date.today, nullable = False, index = True )
thru_date = schema.Column( Date(), default = end_of_times, index = True )
comment = schema.Column( Unicode( 256 ) )
@hybrid.hybrid_property
def mechanism( self ):
if self.contact_mechanism != None:
return self.contact_mechanism.mechanism
@mechanism.setter
def mechanism_setter( self, value ):
if value != None:
if self.contact_mechanism:
self.contact_mechanism.mechanism = value
else:
self.contact_mechanism = ContactMechanism( mechanism = value )
@mechanism.expression
def mechanism_expression( self ):
return sql.select(
[ContactMechanism.mechanism],
whereclause=ContactMechanism.id==self.contact_mechanism_id).as_scalar()
def party_name( self ):
return sql.select( [Party.full_name],
whereclause = (Party.id==self.party_id))
party_name = ColumnProperty( party_name, deferred = True )
def __unicode__( self ):
return six.text_type( self.contact_mechanism )
Admin = PartyContactMechanismAdmin
# begin category definition
class PartyCategory( Entity ):
using_options( tablename = 'party_category' )
name = schema.Column( Unicode(40), index=True, nullable = False )
color = schema.Column( camelot.types.Color() )
# end category definition
parties = ManyToMany( 'Party', lazy = True, backref='categories',
tablename='party_category_party',
remote_colname='party_id',
local_colname='party_category_id')
def get_contact_mechanisms(self, virtual_address_type):
"""Function to be used to do messaging
:param virtual_address_type: a virtual address type, such as 'phone' or 'email'
:return: a generator that yields strings of contact mechanisms, egg 'info@example.com'
"""
for party in self.parties:
for party_contact_mechanism in party.contact_mechanisms:
contact_mechanism = party_contact_mechanism.contact_mechanism
if contact_mechanism:
virtual_address = contact_mechanism.mechanism
if virtual_address and virtual_address[0] == virtual_address_type:
yield virtual_address[1]
def __unicode__(self):
return self.name or ''
class Admin( EntityAdmin ):
verbose_name = _('Category')
verbose_name_plural = _('Categories')
list_display = ['name', 'color']
class PartyAdmin( EntityAdmin ):
verbose_name = _('Party')
verbose_name_plural = _('Parties')
list_display = ['name', 'email', 'phone'] # don't use full name, since it might be None for new objects
list_search = ['full_name']
form_display = ['addresses', 'contact_mechanisms']
form_size = (700, 700)
field_attributes = dict(addresses = {'admin':AddressAdmin},
contact_mechanisms = {'admin':PartyPartyContactMechanismAdmin},
sex = dict( choices = [( u'M', _('male') ), ( u'F', _('female') )], name=_('Gender')),
name = dict( minimal_column_width = 50, name=_('Name')),
note = dict( delegate = delegates.NoteDelegate ),
first_name = {'name': _('First name')},
last_name = {'name': _('Last name')},
social_security_number = {'name': _('Social security number')},
tax_id = {'name': _('Tax registration')},
)
field_attributes.update( Addressable.Admin.field_attributes )
def get_compounding_objects( self, party ):
for party_contact_mechanism in party.contact_mechanisms:
yield party_contact_mechanism
for party_address in party.addresses:
yield party_address
#def flush(self, party):
#from sqlalchemy.orm.session import Session
#session = Session.object_session( party )
#if session:
##
## flush all contact mechanism related objects
##
#objects = [party]
#deleted = ( party in session.deleted )
#for party_contact_mechanism in party.contact_mechanisms:
#if deleted:
#session.delete( party_contact_mechanism )
#objects.extend([ party_contact_mechanism, party_contact_mechanism.contact_mechanism ])
#session.flush( objects )
Party.Admin = PartyAdmin
class OrganizationAdmin( Party.Admin ):
verbose_name = _( 'Organization' )
verbose_name_plural = _( 'Organizations' )
list_display = ['name', 'tax_id', 'email', 'phone', 'fax']
form_display = TabForm( [( _('Basic'), Form( [ WidgetOnlyForm('note'), 'name', 'email',
'phone',
'fax', 'tax_id',
'street1',
'street2',
'city',
'addresses', 'contact_mechanisms'] ) ),
( _('Branding'), Form( ['logo'] ) ),
] )
field_attributes = dict( Party.Admin.field_attributes )
def get_query( self ):
query = super( OrganizationAdmin, self ).get_query()
query = query.options( orm.joinedload('contact_mechanisms') )
return query
Organization.Admin = OrganizationAdmin
class PersonAdmin( Party.Admin ):
verbose_name = _( 'Person' )
verbose_name_plural = _( 'Persons' )
list_display = ['first_name', 'last_name', 'email', 'phone']
form_display = TabForm( [( _('Basic'), Form( [HBoxForm( [ Form( [WidgetOnlyForm('note'),
'first_name',
'last_name',
'sex',
'email',
'phone',
'fax',
'street1',
'street2',
'city',] ),
[WidgetOnlyForm('picture'),
Stretch()],
] ),
'comment', ], scrollbars = False ) ),
( _('Official'), Form( ['birthdate', 'social_security_number', 'passport_number',
'passport_expiry_date', 'addresses', 'contact_mechanisms',], scrollbars = False ) ),
] )
def get_query( self ):
query = super( PersonAdmin, self ).get_query()
query = query.options( orm.joinedload('contact_mechanisms') )
return query
Person.Admin = PersonAdmin