#!/usr/bin/python2.4 # Converts database contents from the Sarge ispmail tutorial to the # schema of the newer Etch tutorial. # Christoph Haas # License: GPL v2 # # You will need to: aptitude install python-sqlalchemy python-mysqldb # Old database: db_sarge = 'driver://username:password@host:port/database' db_etch = 'mysql://root:root2007@localhost:3306/mailserver' #---------- not much to change after this line from sqlalchemy import * from sqlalchemy.ext.assignmapper import assign_mapper from sqlalchemy.ext.sessioncontext import SessionContext from sqlalchemy.orm.mapper import global_extensions ctx = SessionContext(create_session) global_extensions.append(ctx.mapper_extension) # Connect to the databases meta_sarge = BoundMetaData(db_sarge) meta_etch = BoundMetaData(db_etch) # Define Sarge database schema sarge_domains_table = Table('domains', meta_sarge, Column('domain', Unicode, primary_key=True) ) sarge_forwardings_table = Table('forwardings', meta_sarge, Column('source', Unicode, primary_key=True), Column('destination', Unicode) ) sarge_users_table = Table('users', meta_sarge, Column('email', Unicode, primary_key=True), Column('password', Unicode) ) class SargeDomain(object): def __str__(self): return "SargeDomain: %s" % self.domain class SargeForwarding(object): def __str__(self): return "SargeForwarding: %s -> %s" % (self.source, self.destination) class SargeUser(object): def __str__(self): return "SargeUser: %s (%s)" % (self.email, self.password) assign_mapper(ctx, SargeDomain, sarge_domains_table) assign_mapper(ctx, SargeForwarding, sarge_forwardings_table) assign_mapper(ctx, SargeUser, sarge_users_table) # Define Etch database schema etch_domains_table = Table('virtual_domains', meta_etch, Column('id', Integer, primary_key=True), Column('name', Unicode) ) etch_aliases_table = Table('virtual_aliases', meta_etch, Column('id', Integer, primary_key=True), Column('domain_id', Integer, ForeignKey('virtual_domains.id')), Column('source', Unicode), Column('destination', Unicode) ) etch_users_table = Table('virtual_users', meta_etch, Column('id', Integer, primary_key=True), Column('domain_id', Integer, ForeignKey('virtual_domains.id')), Column('user', Unicode), Column('password', Unicode) ) class EtchDomain(object): def __str__(self): return "EtchDomain: %s" % self.name class EtchAlias(object): def __str__(self): return "EtchAlias: %s -> %s" % (self.source, self.destination) class EtchUser(object): def __str__(self): return "EtchUser: %s (%s)" % (self.user, self.password) assign_mapper(ctx, EtchDomain, etch_domains_table) assign_mapper(ctx, EtchAlias, etch_aliases_table, properties = { 'domain' : relation(EtchDomain, backref='aliases') }) assign_mapper(ctx, EtchUser, etch_users_table, properties = { 'domain' : relation(EtchDomain, backref='users') }) # Delete Etch entries print "Flushing tables for Etch..." etch_domains_table.delete().execute() etch_aliases_table.delete().execute() etch_users_table.delete().execute() # Convert domains print "Converting domains from Sarge..." for sarge_domain in SargeDomain.select(): print "- Domain:", sarge_domain.domain etch_domain = EtchDomain() etch_domain.name = sarge_domain.domain #ctx.current.save(etch_domain) # Commit changes ctx.current.flush() # Convert users print "Converting users from Sarge..." for sarge_user in SargeUser.select(): print "- User:", sarge_user.email # Search for domain the user belongs to user, domain = sarge_user.email.split('@',2) belonging_domain = EtchDomain.get_by(name=domain) # Skip orphaned entries if not belonging_domain: print "!!! Ignoring user '%s'. There is no domain '%s'." % (sarge_user.email, domain) continue # Skip entries without passwords if not sarge_user.password: print "!!! Ignoring user '%s'. No password set." % (sarge_user.password) continue etch_user = EtchUser() etch_user.user = user etch_user.password = func.md5(sarge_user.password) belonging_domain.users.append(etch_user) # Convert forwardings print "Converting forwardings from Sarge..." for sarge_alias in SargeForwarding.select(): print "- Forwarding: %s -> %s" % (sarge_alias.source, sarge_alias.destination) if '@' not in sarge_alias.source: print "!!! Ignoring forwarding for '%s'. Not fully qualified." % (sarge_alias.source) continue # Search for domain the user belongs to user, domain = sarge_alias.source.split('@',2) belonging_domain = EtchDomain.get_by(name=domain) # Skip orphaned entries if not belonging_domain: print "!!! Ignoring forwarding for '%s'. There is no domain '%s'." % (sarge_alias.source, domain) continue etch_alias = EtchAlias() etch_alias.source = user etch_alias.destination = sarge_alias.destination belonging_domain.aliases.append(etch_alias) # Commit changes ctx.current.flush()