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 = Column(String(50), nullable=False)
depends_on = Column(Integer)
created = Column(DateTime, default=datetime.datetime.utcnow,
nullable=False)
updated = Column(DateTime, default=datetime.datetime.utcnow,
onupdate=datetime.datetime.utcnow, nullable=False)
task_id = Column(Integer, ForeignKey('task.id'))
--
You received this message because you are subscribed to the Google
Groups "sqlalchemy-alembic" group.
To unsubscribe from this group and stop receiving emails from it, send
an email to sqlalchemy-alembic+unsubscr...@googlegroups.com
<mailto:sqlalchemy-alembic+unsubscr...@googlegroups.com>.
For more options, visit https://groups.google.com/d/optout.
