[Schevo-devel] Re: migrating from SQL to Schevo
Matthew Scott
mscott at goldenspud.com
Thu Dec 8 18:35:57 EST 2005
On 12/8/05, Matthew Scott <mscott at goldenspud.com> wrote:
> For this first post, here is the old schema. Don't laugh too much,
> especially at the 'type' fields that are strings (populated by a
> drop-down in Zope) instead of references to a 'type' table of some
> sort. :)
Now, here is the current rendition of the new schema, complete with
the "import from xyz" transaction (actual company name withheld) that
does all the dirty work.
When I have a bit of time, and as this app evolves, I'll put in some
more commentary as to the decision processes involved.
Here are the steps I followed to import the data:
1. evo db create --app=iminder iminder.db
2. evo nav --pycrust -t iminder.db
3. >>> tx = app.db.t.import_from_xyz()
4. >>> tx.path = 'import'
5. >>> app.db.execute(tx)
"""iminder schema."""
from schevo.schema import *
schevo.schema.prep(locals())
exec(schevo.icon.schema.preamble)
exec(schevo.identity.schema.preamble)
class Address(E.Entity):
type = f.entity('AddressType')
customer = f.entity('Customer')
address = f.unicode()
city = f.unicode()
state = f.unicode()
postal_code = f.unicode()
_plural = 'Addresses'
def __unicode__(self):
return u'%s :: %s' % (self.type, self.customer)
class AddressType(E.Entity):
name = f.unicode()
_key(name)
class Agent(E.Entity):
name = f.unicode()
_key(name)
class Comment(E.Entity):
about = f.entity('Customer', 'TodoItem')
datetime = f.datetime()
text = f.memo()
user = f.entity('SchevoIdentityUser')
def __unicode__(self):
return u'%s @ %s' % (self.about, self.datetime)
class Contact(E.Entity):
type = f.entity('ContactType')
customer = f.entity('Customer')
info = f.unicode()
def __unicode__(self):
return u'%s :: %s' % (self.type, self.customer)
class ContactType(E.Entity):
name = f.unicode()
_key(name)
class Customer(E.Entity):
number = f.integer()
first_name = f.unicode()
last_name = f.unicode()
company_name = f.unicode(required=False)
birthdate = f.date(required=False)
ssn = f.unicode(required=False)
drivers_license = f.unicode(required=False)
def __unicode__(self):
s = u'%s %s' % (self.first_name, self.last_name)
if self.company_name:
s = u'%s (%s)' % (self.company_name, s)
return s
class Policy(E.Entity):
customer = f.entity('Customer')
type = f.entity('PolicyType')
agent = f.entity('Agent', required=False)
provider = f.entity('Provider', required=False)
number = f.unicode()
billing_rate = f.money()
billing_per = f.unicode()
start_date = f.date()
end_date = f.date()
property_coverage = f.money()
liability = f.money()
insured_items = f.memo()
premium = f.money()
_plural = 'Policies'
def __unicode__(self):
return u'%s @ %s :: %s' % (self.type, self.start_date, self.customer)
class PolicyType(E.Entity):
name = f.unicode()
_key(name)
class Provider(E.Entity):
name = f.unicode()
_key(name)
class Sequence(E.Entity):
"""Sequentially increasing numbers for assigning to items."""
next = f.integer()
_hide('t_create', 't_delete', 't_update', 't_increment')
_initial = [
(100000, ),
]
def t_increment(self):
return E.Sequence._Increment(self)
class _Increment(T.Transaction):
"""Increment the sequence and return the previous value."""
def __init__(self, sequence):
T.Transaction.__init__(self)
self.x.sequence = sequence
def _execute(self, db):
seq = self.x.sequence
current = seq.next
tx = seq.t.update(next=current+1)
db.execute(tx)
return current
@classmethod
def x_get(cls):
"""Returns the singleton entity."""
return db.Sequence[1]
@classmethod
def x_next(cls):
"""Returns the next sequence number."""
return db.execute(db.Sequence.x.get().t.increment())
class TodoItem(E.Entity):
type = f.entity('TodoType')
customer = f.entity('Customer')
status = f.entity('TodoStatus')
description = f.unicode()
start_date = f.date()
due_date = f.date()
def __unicode__(self):
return u'%s @ %s :: %s' % (self.type, self.start_date, self.customer)
class TodoStatus(E.Entity):
name = f.unicode()
_key(name)
class TodoType(E.Entity):
name = f.unicode()
_key(name)
class ImportFromXyz(T.Transaction):
path = f.path()
def _execute(self, db):
import os
def records_from_file(name):
# Don't use the csv module, because it can't handle
# embedded newlines quite right.
f = file(os.path.join(self.path, name))
# Compensate for newlines within a record.
lines = []
cur_line = u''
for line in f:
line = line.strip('\n').decode('iso-8859-1')
if line.endswith(u'\\'):
cur_line += line[:-1] + u'\n'
else:
cur_line += line
lines.append(cur_line)
cur_line = u''
return [line.replace(u'\\\\', u'\\').split(u'\t') for line in lines]
# Import customers.
records = records_from_file('customer.txt')
print 'Importing', len(records), 'customers...'
customer_map = {}
for record in records:
(customerid, firstname, lastname, companyname, birthdate,
ssn, driverslicense) = record
tx = db.Customer.t.create()
tx.number = customerid
tx.first_name = firstname
tx.last_name = lastname
tx.company_name = companyname or UNASSIGNED
if birthdate.startswith('0000'):
birthdate = UNASSIGNED
tx.birthdate = birthdate
tx.ssn = ssn or UNASSIGNED
tx.drivers_license = driverslicense or UNASSIGNED
customer_map[customerid] = db.execute(tx)
# Import todo items.
records = records_from_file('todo.txt')
print 'Importing', len(records), 'todo items...'
todo_map = {}
for record in records:
(todoid, customerid, type, status, description, startdate, duedate
) = record
if customerid not in customer_map:
print 'Customer does not exist for', record
continue
tx = db.TodoItem.t.create()
tx.customer = customer_map[customerid]
tx.type = db.execute(db.TodoType.t.create_if_necessary(name=type))
tx.status = db.execute(
db.TodoStatus.t.create_if_necessary(name=status))
tx.description = description
tx.start_date = startdate
tx.due_date = duedate
todo_map[todoid] = db.execute(tx)
# Import todo comments.
records = records_from_file('todocomment.txt')
print 'Importing', len(records), 'todo item comments...'
for record in records:
todocommentid, todoid, commentdate, commenttext, username = record
if todoid not in todo_map:
print 'Todo does not exist for', record
continue
tx = db.Comment.t.create()
tx.about = todo_map[todoid]
tx.text = commenttext
tx.datetime = commentdate
tx.user = db.execute(db.SchevoIdentityUser.t.create_if_necessary(
name=username, password='12345',
))
db.execute(tx)
# Import addresses.
records = records_from_file('address.txt')
print 'Importing', len(records), 'addresses...'
for record in records:
(addressid, addresstype, customerid, address, city,
state, zip) = record
tx = db.Address.t.create()
tx.type = db.execute(
db.AddressType.t.create_if_necessary(name=addresstype))
tx.customer = customer_map[customerid]
tx.address = address
tx.city = city
tx.state = state
tx.postal_code = zip
db.execute(tx)
# Import comments.
records = records_from_file('comment.txt')
print 'Importing', len(records), 'customer comments...'
for record in records:
commentid, customerid, commenttext, commentdate, username = record
tx = db.Comment.t.create()
tx.about = customer_map[customerid]
tx.text = commenttext
tx.datetime = commentdate
tx.user = db.execute(db.SchevoIdentityUser.t.create_if_necessary(
name=username, password='12345',
))
db.execute(tx)
# Import contacts.
records = records_from_file('contact.txt')
print 'Importing', len(records), 'contacts...'
for record in records:
contactid, customerid, contacttype, contactinfo = record
tx = db.Contact.t.create()
tx.customer = customer_map[customerid]
tx.type = db.execute(
db.ContactType.t.create_if_necessary(name=contacttype))
tx.info = contactinfo
db.execute(tx)
# Import policies.
records = records_from_file('policy.txt')
print 'Importing', len(records), 'policies...'
for record in records:
(policyid, customerid, policytype, policynum, billingrate,
billingper, startdate, enddate, propertycoverage, liability,
insureditems, premium, agent, company) = record
tx = db.Policy.t.create()
tx.customer = customer_map[customerid]
tx.type = db.execute(
db.PolicyType.t.create_if_necessary(name=policytype))
tx.number = policynum
tx.billing_rate = billingrate
tx.billing_per = billingper
tx.start_date = startdate
tx.end_date = enddate
tx.property_coverage = propertycoverage
tx.liability = liability
tx.insured_items = insureditems
tx.premium = premium
agent = agent or UNASSIGNED
if agent:
# Normalize the agent name.
agent = u' '.join(agent.strip().title().split())
# Create the agent.
agent = db.execute(
db.Agent.t.create_if_necessary(name=agent))
tx.agent = agent
provider = company or UNASSIGNED
if provider:
provider = db.execute(
db.Provider.t.create_if_necessary(name=provider))
tx.provider = provider
db.execute(tx)
def t_import_from_xyz():
return ImportFromXyz()
--
Matthew R. Scott
More information about the Schevo-devel
mailing list