dbconan_schema.py

8/9/2019 dbconan_schema.py

1/5

File: /home/raka/.cache/.fr-4HLTlV/onan_1.0_src/dbconan_schema.py Page 1

mport datetimemport sys

lass Schema:def __init__(self, dbHost, dbPort, dbName, dbUser, dateOfCreation, description = None):

if dbHost isNone: raise"dbHost can not be None"if dbPort isNone: raise"dbPort can not be None"if dbName isNone: raise"dbName can not be None"if dbUser isNone: raise"dbUser can not be None"if dateOfCreation isNone: raise"dateOfCreation can not be None"ifisinstance(dateOfCreation, datetime.datetime) == False:

raise"dateOfCreation must be of type datetime.datetime"

self.dbHost = dbHostself.dbPort = int(dbPort)self.dbName = dbNameself.dbUser = dbUserself.dateOfCreation = dateOfCreationself.description = description

self.tables = {}

def addTable(self, table):if table isNone: raise"table can not be None"

ifself.tables.has_key(table.name):raise"table %s is already registered in the schema" % table.name

self.tables[table.name] = table

def getTable(self, tableName):if tableName isNone: raise"tableName can not be None"returnself.tables.get(tableName)

def removeTable(self, tableName):if tableName isNone: raise"tableName can not be None"delself.tables[tableName]

lass Table:def __init__(self, name, columnNames, columnTypes, columnKeyNames, pkName, pkColumnNames):

if name isNone: raise"name can not be None"if columnNames isNone: raise"columnNames can not be None"if columnTypes isNone: raise"columnTypes can not be None"if columnKeyNames isNone: raise"columnKeyNames can not be None"iflen(columnNames) < 1: raise"number of columns must be at least 1"iflen(columnNames) != len(columnTypes): raise"len(columnNames) != len(columnTypes)"

#TODO: there shouldn't be duplicate in the columns (the names)columns = []columnPosition = 1for columnName in columnNames:

column = Column(columnName, columnTypes[columnPosition - 1], columnKeyNames[columnPosition ], columnPosition)

columns.append(column)

columnPosition += 1

pk = Noneif pkName isNone:

iflen(pkColumnNames) > 0:raise"pkColumnNames must be zero-length when pkName is None"

else:iflen(pkColumnNames) == 0:

raise"pkColumnNames must not be zero-length when pkName is not None"iflen(pkColumnNames) > len(columnNames):

raise"length pkColumnNames must not be greater than that of columnNames"

pkElements = []pkElementPosition = 1


2/5


for pkColumnName in pkColumnNames:for column in columns:

if column.name == pkColumnName:pkElement = KeyElement(column, pkElementPosition)pkElementPosition += 1pkElements.append(pkElement)break

else:raise"a column named %s can not be found in the table %s" % (pkColumnName, name)

pk = PrimaryKey(self, pkName, pkElements)

self.name = nameself.columns = columnsself.pk = pkself.fks = []self.allPossibleFks = []self.allPossibleFksLocked = False

def isConnected(self):anyOutgoing = len(self.fks) > 0anyIncoming = (self.pk isnotNone) andlen(self.pk.fks) > 0

return anyOutgoing or anyIncoming

def spawnLite(self):columnNames = []columnTypes = []columnKeyNames = []pkColumnNames = []pkName = None

for col inself.columns:columnNames.append(col.name)columnTypes.append(col.type)columnKeyNames.append(col.keyName)

ifself.pk isnotNone:for element inself.pk.elements:

pkColumnNames.append(element.column.name)pkName = self.pk.name

liteTable = Table(self.name, columnNames, columnTypes, columnKeyNames, pkName, pkColumnNames)liteTable.allPossibleFks = self.allPossibleFksliteTable.allPossibleFksLocked = Truereturn liteTable

def referencesFrom(self, referringTable):referringFks = []for fk inself.pk.fks:

if fk.table is referringTable:referringFks.append(fk)

iflen(referringFks) < 1: returnNonereturn referringFks

def referencesTo(self, referredTable):referringFks = []for fk inself.fks:

if fk.pk.table is referredTable:referringFks.append(fk)

iflen(referringFks) < 1: returnNonereturn referringFks

def getColumn(self, name):for col inself.columns:


3/5


if col.name == name:return col

else:returnNone

#forward-lookingdef disconnect(self, fkName):

if fkName isNone: raise"fkName can not be null"

fk = Nonei = 0for existingFk inself.fks:

print existingFk.nameif existingFk.name == fkName:

existingFk.pk.table.unregisterIncomingConnection(existingFk)self.fks.pop(i)break

i += 1else:

raise"fk with name %s can not be found in the table %s" % (fkName, self.name)

#forward-lookingdef connect(self, referredTable, fkName, fkColumnNames):

if referredTable isNone: raise"referredTable can not be null"

if fkColumnNames isNone: raise"fkColumnName can not be null"iflen(fkColumnNames) isNone: raise"fkColumnName must have at least 1 entry"iflen(fkColumnNames) != len(referredTable.pk.elements):

raise"The length of fkColumnNames != referredTable.pk.elements"

fkElements = []fkElementPosition = 1for fkColumnName in fkColumnNames:

for column inself.columns:if column.name == fkColumnName:

correspondingColumn = referredTable.pk.elements[fkElementPosition - 1].column#if column.type != correspondingColumn.type:# raise "The type of column %s in this table (%s) doesn't match the type of

olumn %s in table %s" % (column.name, self.name, referredTable.name,orrespondingColumn.name)

fkElement = KeyElement(column, fkElementPosition)fkElementPosition += 1fkElements.append(fkElement)break

else:raise"a column named %s can not be found in the table %s" % (fkColumnName, name)

fksReferringToReferredTable = self.referencesTo(referredTable)if fksReferringToReferredTable isnotNone:

for fk in fksReferringToReferredTable:if fk.name == fkName:

raise"Foreign key with name %s already exists in this table (%s)" % (fkName,elf.name)

if fk.elements == fkElements:

raise"this table (%s) is already connected to table %s with columns: %s" %self.name, referredTable.name, fkColumnNames)

fk = ForeignKey(self, fkName, fkElements, referredTable.pk)self.fks.append(fk)ifself.allPossibleFksLocked == False:

tableLite = fk.table.spawnLite()tableLite2 = fk.pk.table.spawnLite()fkLite = ForeignKey(tableLite, fk.name, fk.elements, tableLite2.pk)self.allPossibleFks.append(fkLite)

referredTable.registerIncomingConnection(fk)

def unregisterIncomingConnection(self, fkInReferringTable):if fkInReferringTable isNone: raise"fkInReferringTable can not be None"


4/5


ifself.referencesFrom(fkInReferringTable.table) isNone:raise"Table %s is not referring to this table (%s)" % (fkInReferringTable.table.name,

elf.name)

i = 0for referringFk inself.pk.fks:

if fkInReferringTable.name == referringFk.name:self.pk.fks.pop(i)break

i += 1

def registerIncomingConnection(self, fkInReferringTable):if fkInReferringTable isNone: raise"fkInReferringTable can not be None"if fkInReferringTable.table.referencesTo(self) isNone:

raise"Table %s is not referring to this table (%s)" % (fkInReferringTable.table.name,elf.name)

for referringFk inself.pk.fks:if fkInReferringTable.name == referringFk.name:

raise"foreign key with name %s is already registered in this table (%s)" %referringFk.name, self.name)

self.pk.fks.append(fkInReferringTable)

lass Column:def __init__(self, name, type, keyName, position, description=None):if name isNone: raise"name can not be None"iftypeisNone: raise"type can not be None"if position isNone: raise"position can not be None"ifint(position) < 1: raise"position can not be less than 1"

self.name = nameself.type = typeself.keyName = keyNameself.position = int(position)self.description = description

def __cmp__(self, otherColumn):ifisinstance(otherColumn, Column) == False: returnFalse

return otherColumn.name == self.name

def __str__(self):returnself.name

lass KeyElement:def __init__(self, column, position):

self.column = columnself.position = int(position)

def __cmp__(self, otherKeyElement):ifisinstance(otherKeyElement, KeyElement) == False: returnFalsereturn otherKeyElement.column == self.column

def __str__(self):returnself.column.name

lass PrimaryKey:def __init__(self, table, name, elements):

self.table = tableself.name = nameself.elements = elementsself.fks = []


lass ForeignKey:def __init__(self, table, name, elements, pk):

self.table = table


5/5


self.name = nameself.elements = elementsself.pk = pk


dbconan_schema.py

Documents