Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def define_binding(db):
class Author(db.Entity):
public_key = orm.PrimaryKey(database_blob)
authored = orm.Set('ChannelNode')
return Author
status = orm.Optional(str)
message = orm.Optional(str)
file_path = orm.Optional(str, nullable=True)
visualization_path = orm.Optional(str)
# setup relationships
collection = orm.Required(Collection)
catalog_entry = orm.Required(str)
class Providers(db.Entity):
provider = orm.PrimaryKey(str)
username = orm.Required(str)
password = orm.Required(str)
class QuestCatalog(db.Entity):
service_id = orm.PrimaryKey(str)
created_at = orm.Required(datetime, default=datetime.now())
updated_at = orm.Optional(datetime)
metadata = orm.Optional(orm.Json, nullable=True)
geometry = orm.Optional(orm.Json, nullable=True)
def define_binding(db):
class ChannelVote(db.Entity):
"""
This ORM class represents votes cast for a channel. A single instance (row), represents a vote from a single
peer (public key) for a single channel (ChannelMetadata entry, essentially represented by a public_key+id_
pair). To allow only a single vote from the channel, it keeps track of when the vote was cast (vote_date)
and what amount was used locally to bump it (last_amount).
"""
rowid = orm.PrimaryKey(int, size=64, auto=True)
voter = orm.Required("ChannelPeer")
channel = orm.Required("ChannelMetadata", reverse='individual_votes')
orm.composite_key(voter, channel)
last_amount = orm.Optional(float, default=0.0)
vote_date = orm.Optional(datetime, default=datetime.utcnow)
return ChannelVote
def __init__(attr, py_type, *args, **keyargs):
if attr.__class__ is Attribute: raise TypeError("'Attribute' is abstract type")
attr.is_required = isinstance(attr, Required)
attr.is_unique = isinstance(attr, Unique) # Also can be set to True later
attr.is_indexed = attr.is_unique # Also can be set to True later
attr.is_pk = isinstance(attr, PrimaryKey)
if attr.is_pk: attr.pk_offset = 0
else: attr.pk_offset = None
attr.id = next_attr_id()
if not isinstance(py_type, basestring) and not isinstance(py_type, type):
raise TypeError('Incorrect type of attribute: %r' % py_type)
if py_type == 'Entity' or py_type is Entity:
raise TypeError('Cannot link attribute to Entity class. Must use Entity subclass instead')
attr.py_type = py_type
attr.is_collection = isinstance(attr, Collection)
attr.is_ref = not attr.is_collection and isinstance(attr.py_type, (EntityMeta, basestring))
attr.is_basic = not attr.is_collection and not attr.is_ref
attr.sql_type = keyargs.pop('sql_type', None)
attr.entity = attr.name = None
attr.args = args
attr.auto = keyargs.pop('auto', False)
# of uptime.
# b. Repeated votes by some peer to some channel _do not add up_. Instead, the vote is refreshed by substracting
# the old amount from the current vote (it is stored in the DB), and adding the new one (1.0 votes, scaled). This
# is the reason why we have to keep the old votes in the DB, and normalize the old votes last_amount values - to
# keep them in the same "normalization space" to be compatible with the current votes values.
# This binding is used to store normalization data and stats for VSIDS
class Vsids(db.Entity):
"""
This ORM class is used to hold persistent information for the state of VSIDS scoring system.
ACHTUNG! At all times there should be no more than one row/entity of this class. A single entity is
enough to keep the information for the whole GigaChannels.
In a sense, *this is a singleton*.
"""
rowid = orm.PrimaryKey(int)
bump_amount = orm.Required(float)
total_activity = orm.Required(float)
last_bump = orm.Required(datetime)
rescale_threshold = orm.Optional(float, default=10.0 ** 100)
exp_period = orm.Optional(float, default=24.0 * 60 * 60) # decay e times over this period
max_val = orm.Optional(float, default=1.0)
@db_session
def rescale(self, norm):
for channel in db.ChannelMetadata.select(lambda g: g.status != LEGACY_ENTRY):
channel.votes /= norm
for vote in db.ChannelVote.select():
vote.last_amount /= norm
self.max_val /= norm
self.total_activity /= norm
from pony.orm import Database, Required, db_session, Set, Optional, select, \
PrimaryKey, RowNotFound, ERDiagramError, OperationalError
from mutmut import BAD_TIMEOUT, OK_SUSPICIOUS, BAD_SURVIVED, UNTESTED, \
OK_KILLED, RelativeMutationID, Context, mutate
db = Database()
current_db_version = 4
NO_TESTS_FOUND = 'NO TESTS FOUND'
class MiscData(db.Entity):
key = PrimaryKey(str, auto=True)
value = Optional(str, autostrip=False)
class SourceFile(db.Entity):
filename = Required(str, autostrip=False)
hash = Optional(str)
lines = Set('Line')
class Line(db.Entity):
sourcefile = Required(SourceFile)
line = Optional(str, autostrip=False)
line_number = Required(int)
mutants = Set('Mutant')
import pony.orm as orm
import shutil
import uuid
db = orm.Database(**config.database)
session = orm.db_session
commit = orm.commit
rollback = orm.rollback
select = orm.select
desc = orm.desc
class User(db.Entity):
_table_ = 'users'
id = orm.PrimaryKey(int)
name = orm.Required(str, unique=True)
passhash = orm.Required(str)
can_manage = orm.Required(bool)
can_download_artifacts = orm.Required(bool)
can_view_buildlogs = orm.Required(bool)
login_tokens = orm.Set('LoginToken')
def __init__(self, **kwargs):
if 'id' not in kwargs:
kwargs['id'] = (orm.max(x.id for x in User) or 0) + 1
super().__init__(**kwargs)
def set_password(self, password):
self.passhash = utils.hash_pw(password)
@classmethod
def define_binding(db):
class MiscData(db.Entity):
"""
This binding is used to store all kinds of values, like DB version, counters, etc.
"""
name = orm.PrimaryKey(str)
value = orm.Optional(str)
return MiscData
db.bind(provider="mysql", host=creds['host'], user=creds['username'], password=creds['password'],
db=creds['database'], ssl=ssl, port=int(creds.get('port', 3306)))
else:
raise Exception("No database configuration")
db.generate_mapping(create_tables=True)
class GifHosts(db.Entity):
name = PrimaryKey(str)
origin_gifs = Set('Gif', reverse='origin_host')
reversed_gifs = Set('Gif', reverse='reversed_host')
class Gif(db.Entity):
id = PrimaryKey(int, auto=True)
origin_host = Required(GifHosts, reverse='origin_gifs')
origin_id = Required(str)
reversed_host = Required(GifHosts, reverse='reversed_gifs')
reversed_id = Required(str)
time = Required(date)
nsfw = Optional(bool)
total_requests = Optional(int)
last_requested_date = Optional(date)
bind_db(db)
ghm = GifHostManager()
def sync_hosts():
# Double check gifhost bindings