Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class FancySellable:
"""A fancy class used by some kiwi entries."""
# XXX Probably we could avoid this class with some kiwi improvements
# waiting for bug 2365.
def __init__(self, price=0.0, quantity=1.0, unit=None):
self.price = price
self.quantity = quantity
self.unit = unit
def get_unit_description(self):
return self.unit and self.unit.description or ""
class AbstractSellableCategory(Domain):
description = StringCol()
suggested_markup = FloatCol(default=0.0)
# A percentage commission suggested for all the sales which products
# belongs to this category or base category
salesperson_commission = FloatCol(default=0.0)
def get_commission(self):
return self.salesperson_commission
class BaseSellableCategory(Domain):
category_data = ForeignKey('AbstractSellableCategory')
implements(IDescribable)
def get_commission(self):
STATUS_REVIEWING: _('Reviewing'),
STATUS_CONFIRMED: _('Confirmed'),
STATUS_CANCELLED: _('Cancelled')}
# XXX The payment_id attribute will be an alternateID after
# fixing bug 2214
payment_id = IntCol(default=None)
status = IntCol(default=STATUS_PREVIEW)
due_date = DateTimeCol()
paid_date = DateTimeCol(default=None)
paid_value = FloatCol(default=0.0)
base_value = FloatCol()
value = FloatCol()
interest = FloatCol(default=0.0)
discount = FloatCol(default=0.0)
description = StringCol(default=None)
payment_number = StringCol(default=None)
method = ForeignKey('AbstractPaymentMethodAdapter')
method_details = ForeignKey('PaymentMethodDetails', default=None)
group = ForeignKey('AbstractPaymentGroup')
destination = ForeignKey('PaymentDestination')
#
# SQLObject hooks
#
def _create(self, id, **kw):
if not 'value' in kw:
raise TypeError('You must provide a value argument')
if not 'base_value' in kw or not kw['base_value']:
import datetime
import gtk
from sqlobject import connectionForURI, SQLObject, StringCol, DateCol, ForeignKey, AND
from kiwi.enums import SearchFilterPosition
from kiwi.db.sqlobj import SQLObjectQueryExecuter
from kiwi.ui.objectlist import Column
from kiwi.ui.search import (SearchContainer, DateSearchFilter,
ComboSearchFilter)
__connection__ = connectionForURI('sqlite:///:memory:')
class Category(SQLObject):
name = StringCol()
Category.createTable()
class Task(SQLObject):
title = StringCol()
category = ForeignKey('Category')
date = DateCol()
Task.createTable()
for category in ['Work',
'Home',
'School']:
Category(name=category)
work = Category.selectBy(name='Work')[0]
home = Category.selectBy(name='Home')[0]
"""
name = StringCol()
short_name = StringCol()
compensation_code = StringCol()
class BankAccount(Domain):
"""A bank account definition.
B{Important atributes}:
- I{bank_id}: the bank identifier.
- I{branch}: the bank branch where this account lives.
- I{account}: an identifier of this account in the branch.
"""
bank_id = IntCol(default=0)
branch = StringCol(default=None)
account = StringCol(default=None)
from kiwi.enums import SearchFilterPosition
from kiwi.db.sqlobj import SQLObjectQueryExecuter
from kiwi.ui.objectlist import Column
from kiwi.ui.search import (SearchContainer, DateSearchFilter,
ComboSearchFilter)
__connection__ = connectionForURI('sqlite:///:memory:')
class Category(SQLObject):
name = StringCol()
Category.createTable()
class Task(SQLObject):
title = StringCol()
category = ForeignKey('Category')
date = DateCol()
Task.createTable()
for category in ['Work',
'Home',
'School']:
Category(name=category)
work = Category.selectBy(name='Work')[0]
home = Category.selectBy(name='Home')[0]
school = Category.selectBy(name='School')[0]
today = datetime.date.today()
for title, category, date in [
('Upgrade web server', work, today - datetime.timedelta(1)),
from application import log
from application.python import Null
from application.python.types import Singleton
from sqlobject import connectionForURI, sqlhub, IntCol, SQLObject, StringCol
from twisted.internet.threads import deferToThread
def defer_to_thread(func):
"""Decorator to run DB queries in Twisted's thread pool"""
def wrapper(*args, **kw):
return deferToThread(func, *args, **kw)
return wrapper
class Users(SQLObject):
username = StringCol()
class UserKeys(SQLObject):
user_id = IntCol()
key = StringCol(sqlType='LONGTEXT')
class DatabaseError(Exception): pass
class Database(object):
__metaclass__ = Singleton
def __init__(self, dburi):
if ':memory:' in dburi:
log.warn('SQLite in-memory DB is not supported')
dburi = None
#
class PaymentDestination(InheritableModel):
"""PaymentDestination is the location where all the paid payments live.
B{Important attributes}:
- I{description}: an easy identification for this payment
destination.
- I{account}: if this payment destination represents a bank account,
use it here.
"""
implements(IDescribable)
description = StringCol()
account = ForeignKey('BankAccount', default=None)
notes = StringCol(default='')
def get_balance(self, start_date=None, end_date=None):
raise NotImplementedError
#
# IDescribable implementation
#
def get_description(self):
return self.description
class StoreDestination(PaymentDestination):
"""A StoreDestination is a payment destination which lives in a Store.
Most of times this will represent the total value of operations in this
store.
# URL of the page
description = StringCol(default=None)
# An ad must be unique on the website
adIndex = DatabaseIndex('siteId', 'site', unique=True)
class Rent(SQLObject):
# Total cost
totalRent = FloatCol(default=None)
# Surface in square meters
livingSpace = FloatCol(default=None)
# Currency, 3 letter code
currency = StringCol(length=3, default='EUR')
# Date of arrival
moveIn = DateCol(default=None)
# IP of the user hashed
ipHash = StringCol(length=512,default=None)
# Date of creating
createdAt = DateTimeCol(default=DateTimeCol.now)
class Report(SQLObject):
createdAt = DateTimeCol(default=DateTimeCol.now) # Date the ad was first scraped
country = StringCol(length=2, default=None) # Country, 2 letter code
site = StringCol(length=30, default=None) # Name of the website
siteId = StringCol(length=100, default=None) # The unique ID from the site where it's scrapped from
name = StringCol(length=100, default=None) # Name of the report: duplicate, bogus, rent-missing, space-missing, timeout
if not DatabaseConfig.dburi:
raise RuntimeError('Database accounting is enabled, but the database URI is not specified in config.ini')
connection = connectionForURI(DatabaseConfig.dburi)
sqlhub.processConnection = connection
class MediaSessions(SQLObject):
class sqlmeta:
table = DatabaseConfig.sessions_table
createSQL = {'mysql': 'ALTER TABLE %s ENGINE MyISAM' % DatabaseConfig.sessions_table}
cacheValues = False
call_id = StringCol(length=255, dbName=DatabaseConfig.callid_column, notNone=True)
from_tag = StringCol(length=64, dbName=DatabaseConfig.fromtag_column, notNone=True)
to_tag = StringCol(length=64, dbName=DatabaseConfig.totag_column)
info = BLOBCol(length=2**24-1, dbName=DatabaseConfig.info_column) # 2**24-1 makes it a mediumblob in mysql, that can hold 16 million bytes
# Indexes
callid_idx = DatabaseIndex('call_id', 'from_tag', 'to_tag', unique=True)
try:
MediaSessions.createTable(ifNotExists=True)
except OperationalError as e:
log.error("cannot create the `%s' table: %s" % (DatabaseConfig.sessions_table, e))
log.info("please make sure that the `%s' user has the CREATE and ALTER rights on the `%s' database" % (connection.user, connection.db))
log.info('then restart the dispatcher, or you can create the table yourself using the following definition:')
log.info('----------------- >8 -----------------')
sql, constraints = MediaSessions.createTableSQL()
statements = ';\n'.join([sql] + constraints) + ';'
log.info(statements)