Done, new file attached (this gives the same error message as the one I
showed initially, at least on my system).
- Ken
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
from sqlalchemy import *
from sqlalchemy.dialects.mysql import INTEGER, SMALLINT, TIMESTAMP
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
import datetime
Base = declarative_base()
class AppDefinition(Base):
__tablename__ = 'app_definitions'
id = Column(u'AppID', SMALLINT(display_width=2), primary_key=True)
app_type = Column(u'appType', String(length=100), nullable=False)
status = Column(
Enum('active', 'inactive'),
nullable=False,
server_default='active'
)
app_deployments = relationship('AppDeployment')
class AppDeployment(Base):
__tablename__ = 'app_deployments'
id = Column(u'AppDeploymentID', INTEGER(), primary_key=True)
deployment_id = Column(
u'DeploymentID',
INTEGER(),
ForeignKey('deployments.DeploymentID', ondelete='cascade'),
nullable=False
)
app_id = Column(
u'AppID',
SMALLINT(display_width=6),
ForeignKey('app_definitions.AppID', ondelete='cascade'),
nullable=False
)
user = Column(String(length=32), nullable=False)
status = Column(
Enum(
'complete',
'incomplete',
'inprogress',
'invalidated',
'validated',
),
nullable=False
)
environment_id = Column(
u'environment_id',
INTEGER(),
ForeignKey('environments.environmentID', ondelete='cascade'),
nullable=False
)
realized = Column(
TIMESTAMP(),
nullable=False,
server_default=func.current_timestamp()
)
application = relationship('AppDefinition')
deployment = relationship('Deployment')
environment_obj = relationship('Environment')
@hybrid_property
def environment(self):
return self.environment_obj.environment
@environment.expression
def environment(cls):
return select(
[Environment.environment]
).where(
Environment.id == cls.environment_id
).correlate(cls).as_scalar()
class Deployment(Base):
__tablename__ = 'deployments'
id = Column(u'DeploymentID', INTEGER(), primary_key=True)
package_id = Column(
INTEGER(),
ForeignKey('packages.package_id', ondelete='cascade'),
nullable=False
)
app_deployments = relationship('AppDeployment')
class Environment(Base):
__tablename__ = 'environments'
id = Column(u'environmentID', INTEGER(), primary_key=True)
environment = Column(String(length=15), nullable=False, unique=True)
class Package(Base):
__tablename__ = 'packages'
id = Column(u'package_id', INTEGER(), primary_key=True)
pkg_name = Column(String(length=255), nullable=False)
version = Column(String(length=63), nullable=False)
revision = Column(String(length=63), nullable=False)
deployments = relationship('Deployment')
Session = scoped_session(sessionmaker())
subq = (Session.query(Package.pkg_name, Package.version,
Package.revision, AppDefinition.app_type,
AppDeployment.environment,
AppDeployment.realized, AppDeployment.user,
AppDeployment.status)
.join(Deployment)
.join(AppDeployment)
.join(AppDefinition)
.filter(AppDeployment.status != 'invalidated')
.filter(AppDeployment.environment == "some environment")
.order_by(AppDeployment.realized.desc())
.subquery(name='t_ordered'))
q = (Session.query(subq.c.pkg_name, subq.c.version, subq.c.revision,
subq.c.appType, subq.c.environment,
subq.c.realized, subq.c.user, subq.c.status)
.group_by(subq.c.appType, subq.c.environment_id,
subq.c.pkg_name)
.having(and_(subq.c.status.like('%complete'),
func.unix_timestamp(subq.c.realized) <
func.unix_timestamp(func.now()) -
datetime.timedelta(seconds=60)))
)
print q
