Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import datetime
import sqlalchemy as sa
import sqlalchemy.orm as orm
from pypi.data.modelbase import SqlAlchemyBase
from pypi.data.releases import Release
class Package(SqlAlchemyBase):
__tablename__ = 'packages'
id = sa.Column(sa.String, primary_key=True)
created_date = sa.Column(sa.DateTime, default=datetime.datetime.now)
last_updated = sa.Column(sa.DateTime, default=datetime.datetime.now)
summary = sa.Column(sa.String, nullable=True)
description = sa.Column(sa.String, nullable=True)
home_page = sa.Column(sa.String, nullable=True)
docs_url = sa.Column(sa.String, nullable=True)
package_url = sa.Column(sa.String, nullable=True)
author_name = sa.Column(sa.String, nullable=True)
author_email = sa.Column(sa.String, nullable=True, index=True)
license = sa.Column(sa.String, index=True)
releases = orm.relation("Release", order_by=[
Release.major_ver.desc(),
Release.minor_ver.desc(),
Release.build_ver.desc(),
], back_populates='package')
def setup(self):
self.engine = engines.mock_engine()
self.metadata = MetaData(self.engine)
self.users = Table(
"users",
self.metadata,
Column("user_id", Integer, primary_key=True),
Column("user_name", String(40)),
)
def test_model_out_of_context(self):
class Foo(self.db.Model):
__tablename__ = 'foo'
id = Column(Integer, primary_key=True)
name = Column(String)
with self.db.session() as session:
session.execute('create table foo(id int primary key, name text);')
session.execute("insert into foo values(1, 'foo')")
foo = session.query(Foo).first()
self.assertEqual(1, foo.id)
def upgrade():
op.create_table('css_files',
sa.Column('created_dt', UTCDateTime, nullable=False),
sa.Column('filename', sa.String(), nullable=False),
sa.Column('content_type', sa.String(), nullable=False),
sa.Column('size', sa.BigInteger(), nullable=False),
sa.Column('storage_backend', sa.String(), nullable=False),
sa.Column('storage_file_id', sa.String(), nullable=False),
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('event_id', sa.Integer(), nullable=False, unique=True, index=True),
sa.ForeignKeyConstraint(['event_id'], ['events.events.id']),
sa.PrimaryKeyConstraint('id'),
schema='events')
import opentracing
import sqlalchemy_opentracing
DB_LOCATION = '/tmp/simple.db'
# Your OpenTracing-compatible tracer here.
tracer = opentracing.Tracer()
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
if __name__ == '__main__':
if os.path.exists(DB_LOCATION):
os.remove(DB_LOCATION) # cleanup
engine = create_engine('sqlite:///%s' % DB_LOCATION)
session = sessionmaker(bind=engine)()
sqlalchemy_opentracing.init_tracing(tracer)
sqlalchemy_opentracing.register_engine(engine)
User.metadata.create_all(engine)
# Register the session for the current transaction.
sqlalchemy_opentracing.set_traced(session)
"""Retrieve a partition by id, returns the partition or None if no partition
with that id exists."""
return Partition.query.filter_by(image_id=image_id, table=table, slot=slot).first()
@staticmethod
def add(part):
"""Add a new partition to the datatbase."""
_add(part)
class FileElement(BASE):
"""
Class to hold basic details of a ByteSequence, used in file analysis.
"""
__tablename__ = 'file_element'
id = Column(Integer, primary_key=True)# pylint: disable-msg=C0103
path = Column(String(4096), nullable=False)
image_id = Column(String(10), nullable=False)
partition_table = Column(Integer, nullable=False)
partition_slot = Column(Integer, nullable=False)
partition = relationship('Partition', backref=backref('file_elements', lazy='dynamic'))
byte_sequence_id = Column(Integer, ForeignKey('byte_sequence.id'), nullable=False)
byte_sequence = relationship('ByteSequence', backref=backref('file_elements', lazy='dynamic'))
__table_args__ = (UniqueConstraint('image_id', 'partition_table', 'partition_slot',
'path', name='uix_partition_path'),
ForeignKeyConstraint([image_id, partition_table, partition_slot],
[Partition.image_id, Partition.table, Partition.slot]),
{})
def __init__(self, path, partition, byte_sequence):
sa.Column('is_removed', sa.Boolean(), nullable=False),
sa.Column('is_auto_created', sa.Boolean(), nullable=False),
sa.PrimaryKeyConstraint('id', name=op.f('pk_tasks'))
)
op.create_index(op.f('ix_tasks_entry_name'), 'tasks', ['entry_name'], unique=False)
op.create_index(op.f('ix_tasks_is_auto_created'), 'tasks', ['is_auto_created'], unique=False)
op.create_index(op.f('ix_tasks_is_removed'), 'tasks', ['is_removed'], unique=False)
op.create_index(op.f('ix_tasks_task_name'), 'tasks', ['task_name'], unique=False)
op.create_table('taskexecutions',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('task_id', sa.Integer(), nullable=False),
sa.Column('planned_at', TIMESTAMP(timezone=True), nullable=False),
sa.Column('locked_at', TIMESTAMP(timezone=True), nullable=True),
sa.Column('finished_at', TIMESTAMP(timezone=True), nullable=True),
sa.Column('canceled_at', TIMESTAMP(timezone=True), nullable=True),
sa.Column('log', sa.String(), nullable=True),
sa.ForeignKeyConstraint(['task_id'], ['tasks.id'], name=op.f('fk_taskexecutions_task_id_tasks'), ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_taskexecutions'))
)
op.create_index(op.f('ix_taskexecutions_canceled_at'), 'taskexecutions', ['canceled_at'], unique=False)
op.create_index(op.f('ix_taskexecutions_finished_at'), 'taskexecutions', ['finished_at'], unique=False)
op.create_index(op.f('ix_taskexecutions_locked_at'), 'taskexecutions', ['locked_at'], unique=False)
op.create_index(op.f('ix_taskexecutions_planned_at'), 'taskexecutions', ['planned_at'], unique=False)
op.create_index(op.f('ix_taskexecutions_task_id'), 'taskexecutions', ['task_id'], unique=False)
### end Alembic commands ###
location = relationship('Location')
t_active_blockages = Table(
'active_blockages', metadata,
Column('id', Integer),
Column('active', Boolean)
)
class Activegroup(Base):
__tablename__ = 'activegroups'
id = Column(Integer, primary_key=True, server_default=text("nextval('activegroups_id_seq'::regclass)"))
name = Column(String(16), nullable=False)
comment = Column(String(128), nullable=False)
class Active(Base):
__tablename__ = 'actives'
id = Column(Integer, primary_key=True, server_default=text("nextval('actives_id_seq'::regclass)"))
account_id = Column(ForeignKey('accounts.id'), nullable=False, unique=True)
comment = Column(String(128), nullable=False)
activegroup_id = Column(ForeignKey('activegroups.id'), nullable=False, index=True)
account = relationship('Account', uselist=False)
activegroup = relationship('Activegroup')
class Banktransfer(Base):
__tablename__ = 'banktransfers'
def upgrade():
### commands auto generated by Alembic - please adjust! ###
op.create_table('eve_user',
sa.Column('character_id', sa.Integer(), autoincrement=False, nullable=False),
sa.Column('character_owner_hash', sa.String(length=255), nullable=True),
sa.Column('character_name', sa.String(length=200), nullable=True),
sa.Column('scopes', sa.String(length=200), nullable=True),
sa.Column('token_type', sa.String(length=20), nullable=True),
sa.Column('access_token', sa.String(length=100), nullable=True),
sa.Column('access_token_expires_on', sa.DateTime(timezone=True), nullable=True),
sa.Column('access_token_expires_in', sa.Integer(), nullable=True),
sa.Column('refresh_token', sa.String(length=100), nullable=True),
sa.Column('refresh_token_expires_on', sa.DateTime(timezone=True), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text(u'now()'), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text(u'now()'), nullable=True),
sa.PrimaryKeyConstraint('character_id')
)
Column("name", String(200), default='', server_default=''),
Column("gender", INTEGER, default=0, server_default='0'),
Column("birthday", String(200), default='', server_default=''),
Column("location", String(100), default='', server_default=''),
Column("description", String(500), default='', server_default=''),
Column("register_time", String(200), default='', server_default=''),
Column("verify_type", INTEGER, default=0, server_default='0'),
Column("verify_info", String(2500), default='', server_default=''),
Column("follows_num", INTEGER, default=0, server_default='0'),
Column("fans_num", INTEGER, default=0, server_default='0'),
Column("wb_num", INTEGER, default=0, server_default='0'),
Column("level", INTEGER, default=0, server_default='0'),
Column("tags", String(500), default='', server_default=''),
Column("work_info", String(500), default='', server_default=''),
Column("contact_info", String(300), default='', server_default=''),
Column("education_info", String(300), default='', server_default=''),
Column("head_img", String(500), default='', server_default=''),
)
# seed ids for user crawling
seed_ids = Table('seed_ids', metadata,
Column("id", INTEGER, primary_key=True, autoincrement=True),
Column("uid", String(20), unique=True),
Column("is_crawled", INTEGER, default=0, server_default='0'),
Column("other_crawled", INTEGER, default=0, server_default='0'),
Column("home_crawled", INTEGER, default=0, server_default='0'),
)
# search keywords table
keywords = Table('keywords', metadata,
Column("id", INTEGER, primary_key=True, autoincrement=True),
Column("keyword", String(200), unique=True),