Re: MySQL ForeignKey indexes being removed
MyISAM does not actually record foreign key definitions. Autogenerate
is not compatible with foreign keys that are declared in the model yet
are not actually present in the database due to MyISAM. To keep these
FK definitions in Python but work around the autogenerate issue, create
an include_object rule that skips foreign_key_constraint detection:
http://alembic.zzzcomputing.com/en/latest/api/runtime.html?highlight=include_object#alembic.runtime.environment.EnvironmentContext.configure.params.include_object
On 12/22/2016 06:08 PM, Adam Patt wrote:
I'm using mysql 5.1.73 and alembic 0.8.6 and even when the model has not
changed. Below I have the schema migration which is getting auto
generated as well as the models defs. This is using MyISAM and when I
look at the Foreign Keys tab in MySQL Workbench, it says foreign keys
can only be used in certain engines. I only recently added the Foreign
Key and migrations worked exactly as expected before this change.
What am I doing wrong?
### generated schema migration
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.drop_index('fk_command_task_id_task', table_name='command')
op.create_foreign_key(op.f('fk_command_task_id_task'), 'command',
'task', ['task_id'], ['id'])
op.drop_index('fk_task_job_id_job', table_name='task')
op.create_foreign_key(op.f('fk_task_job_id_job'), 'task', 'job',
['job_id'], ['id'])
### end Alembic commands ###
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(op.f('fk_task_job_id_job'), 'task',
type_='foreignkey')
op.create_index('fk_task_job_id_job', 'task', ['job_id'], unique=False)
op.drop_constraint(op.f('fk_command_task_id_task'), 'command',
type_='foreignkey')
op.create_index('fk_command_task_id_task', 'command', ['task_id'],
unique=False)
### end Alembic commands ###
### object definitions
Base = declarative_base()
# make it so constraints have a name so that backward migrations work
correctly
Base.metadata = MetaData(naming_convention={
"ix": 'ix_%(column_0_label)s',
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s"
})
class SnapBase(object):
"""
base for all SNAP sqlalchemy objects
does the following things:
- adds a default id field to all objects
- automatically generate a table name based on the class name.
Turns ClassName to class_name
- creates a nice human readable version of any object
- utility method to get the list of fields available
"""
@declared_attr
def __tablename__(cls):
"""
turn things like MyTable into my_table in the database
"""
tablename = ''
for c in cls.__name__:
if c.isupper() and tablename:
tablename += '_'
tablename += c.lower()
return tablename
def _field_names(self):
return [str(c).split('.')[1] for c in self.__table__.columns]
def __repr__(self):
display = ""
for column in self.__table__.columns:
name = str(column).split('.')[1]
if display:
display += ', '
value = getattr(self, name)
column_type = str(column.type)
if (column_type.startswith('VARCHAR') or
column_type.startswith('CHAR') or column_type in ['DATETIME', 'DATE'])
and value is not None:
val = "%s='%s'"
else:
val = "%s=%s"
display += val % (name, value)
return '%s(%s)' % (self.__class__.__name__, display)
id = Column(Integer, primary_key=True)
class Job(SnapBase, Base):
user = Column(String(50), nullable=False)
status = Column(String(10), default='running', nullable=False) #
running/complete
created = Column(DateTime, default=datetime.datetime.utcnow,
nullable=False)
updated = Column(DateTime, default=datetime.datetime.utcnow,
onupdate=datetime.datetime.utcnow, nullable=False)
tasks = relationship('Task', backref='job')
class Task(SnapBase, Base):
serial = Column(String(50), nullable=False)
name = Column(String(50), nullable=False)
user = Column(String(50), nullable=False)
status = Column(String(10), default='running', nullable=False) #
running/complete
result = Column(String(50))
message = Column(Text)
created = Column(DateTime, default=datetime.datetime.utcnow,
nullable=False)
updated = Column(DateTime, default=datetime.datetime.utcnow,
onupdate=datetime.datetime.utcnow, nullable=False)
job_id = Column(Integer, ForeignKey('job.id'))
commands = relationship('Command', backref='task')
class Command(SnapBase, Base):
serial = Column(String(50), nullable=False)
agent = Column(String(20), nullable=False) # snap/oob/cobbler/chef
MySQL ForeignKey indexes being removed
I'm using mysql 5.1.73 and alembic 0.8.6 and even when the model has not
changed. Below I have the schema migration which is getting auto generated
as well as the models defs. This is using MyISAM and when I look at the
Foreign Keys tab in MySQL Workbench, it says foreign keys can only be used
in certain engines. I only recently added the Foreign Key and migrations
worked exactly as expected before this change.
What am I doing wrong?
### generated schema migration
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.drop_index('fk_command_task_id_task', table_name='command')
op.create_foreign_key(op.f('fk_command_task_id_task'), 'command',
'task', ['task_id'], ['id'])
op.drop_index('fk_task_job_id_job', table_name='task')
op.create_foreign_key(op.f('fk_task_job_id_job'), 'task', 'job',
['job_id'], ['id'])
### end Alembic commands ###
def downgrade():
### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(op.f('fk_task_job_id_job'), 'task',
type_='foreignkey')
op.create_index('fk_task_job_id_job', 'task', ['job_id'], unique=False)
op.drop_constraint(op.f('fk_command_task_id_task'), 'command',
type_='foreignkey')
op.create_index('fk_command_task_id_task', 'command', ['task_id'],
unique=False)
### end Alembic commands ###
### object definitions
Base = declarative_base()
# make it so constraints have a name so that backward migrations work
correctly
Base.metadata = MetaData(naming_convention={
"ix": 'ix_%(column_0_label)s',
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s"
})
class SnapBase(object):
"""
base for all SNAP sqlalchemy objects
does the following things:
- adds a default id field to all objects
- automatically generate a table name based on the class name.
Turns ClassName to class_name
- creates a nice human readable version of any object
- utility method to get the list of fields available
"""
@declared_attr
def __tablename__(cls):
"""
turn things like MyTable into my_table in the database
"""
tablename = ''
for c in cls.__name__:
if c.isupper() and tablename:
tablename += '_'
tablename += c.lower()
return tablename
def _field_names(self):
return [str(c).split('.')[1] for c in self.__table__.columns]
def __repr__(self):
display = ""
for column in self.__table__.columns:
name = str(column).split('.')[1]
if display:
display += ', '
value = getattr(self, name)
column_type = str(column.type)
if (column_type.startswith('VARCHAR') or
column_type.startswith('CHAR') or column_type in ['DATETIME', 'DATE']) and
value is not None:
val = "%s='%s'"
else:
val = "%s=%s"
display += val % (name, value)
return '%s(%s)' % (self.__class__.__name__, display)
id = Column(Integer, primary_key=True)
class Job(SnapBase, Base):
user = Column(String(50), nullable=False)
status = Column(String(10), default='running', nullable=False) #
running/complete
created = Column(DateTime, default=datetime.datetime.utcnow,
nullable=False)
updated = Column(DateTime, default=datetime.datetime.utcnow,
onupdate=datetime.datetime.utcnow, nullable=False)
tasks = relationship('Task', backref='job')
class Task(SnapBase, Base):
serial = Column(String(50), nullable=False)
name = Column(String(50), nullable=False)
user = Column(String(50), nullable=False)
status = Column(String(10), default='running', nullable=False) #
running/complete
result = Column(String(50))
message = Column(Text)
created = Column(DateTime, default=datetime.datetime.utcnow,
nullable=False)
updated = Column(DateTime, default=datetime.datetime.utcnow,
onupdate=datetime.datetime.utcnow, nullable=False)
job_id = Column(Integer, ForeignKey('job.id'))
commands = relationship('Command', backref='task')
class Command(SnapBase, Base):
serial = Column(String(50), nullable=False)
agent = Column(String(20), nullable=False) # snap/oob/cobbler/chef
silo = Column(String(20), nullable=False)
command = Column(String(30), nullable=False)
delay = Column(Integer, default=0, nullable=False)
retries = Column(Integer, default=0, nullable=False)
timeout = Column(Integer, nullable=False)
expire = Column(DateTime, nullable=False)
parameters = Column(JSONEncodedDict(2000), default={}, nullable=False)
# json
status = Column(String(10), default='pending', nullable=False) #
pending, sent, success/error
result = Column(String(50))
message = Column(Text)
user
