def _create_table_version(cls, engine, repository, version):
"""
Creates the versioning table in a database.
:raises: :exc:`DatabaseAlreadyControlledError`
"""
# Create tables
tname = repository.version_table
meta = MetaData(engine)
table = Table(
tname, meta,
Column('repository_id', String(250), primary_key=True),
Column('repository_path', Text),
Column('version', Integer), )
# there can be multiple repositories/schemas in the same db
if not table.exists():
table.create()
# test for existing repository_id
s = table.select(table.c.repository_id == bindparam("repository_id"))
result = engine.execute(s, repository_id=repository.id)
if result.fetchone():
raise exceptions.DatabaseAlreadyControlledError
# Insert data
engine.execute(table.insert().values(
repository_id=repository.id,
repository_path=repository.path,
version=int(version)))
return table
python类bindparam()的实例源码
def init(self, session, api):
self.session = session
self.api = api
# compile queries
self.command_queries = {}
self.command_queries['character'] = \
session.query(Command).\
filter(Command.actor_id == bindparam('actor_id'))
self.command_queries['children'] = \
session.query(Actor, Command).\
filter(Actor.id == bindparam('actor_id')).\
join(Actor.parent, aliased=True).\
filter(Command.actor_id == Actor.id)
location = aliased(Actor)
self.command_queries['location'] = \
session.query(Command).\
join(location).\
join(Actor, location.id == Actor.parent_id).\
filter(Actor.id == bindparam('actor_id'))
location = aliased(Actor)
children = aliased(Actor)
self.command_queries['location_children'] = \
session.query(Command).\
join(children).\
join(location, location.id == children.parent_id).\
join(Actor, location.id == Actor.parent_id).\
filter(Actor.id == bindparam('actor_id')).\
filter(Command.actor_id == children.id)
self.command_precedence = ['character', 'children', 'location', 'location_children']
def _create_table_version(cls, engine, repository, version):
"""
Creates the versioning table in a database.
:raises: :exc:`DatabaseAlreadyControlledError`
"""
# Create tables
tname = repository.version_table
meta = MetaData(engine)
table = Table(
tname, meta,
Column('repository_id', String(250), primary_key=True),
Column('repository_path', Text),
Column('version', Integer), )
# there can be multiple repositories/schemas in the same db
if not table.exists():
table.create()
# test for existing repository_id
s = table.select(table.c.repository_id == bindparam("repository_id"))
result = engine.execute(s, repository_id=repository.id)
if result.fetchone():
raise exceptions.DatabaseAlreadyControlledError
# Insert data
engine.execute(table.insert().values(
repository_id=repository.id,
repository_path=repository.path,
version=int(version)))
return table
