Python sqlalchemy 模块，exists() 实例源码

def _get_or_create_event_type(self, event_type, session):
        """Check if an event type with the supplied name is already exists.

        If not, we create it and return the record. This may result in a flush.
        """
        try:
            with session.begin(nested=True):
                et = session.query(models.EventType).filter(
                    models.EventType.desc == event_type).first()
                if not et:
                    et = models.EventType(event_type)
                    session.add(et)
        except dbexc.DBDuplicateEntry:
            et = self._get_or_create_event_type(event_type, session)

        return et

def add_sighting(session, pokemon):
    # Check if there isn't the same entry already
    if pokemon in SIGHTING_CACHE:
        return
    if session.query(exists().where(and_(
                Sighting.expire_timestamp == pokemon['expire_timestamp'],
                Sighting.encounter_id == pokemon['encounter_id']))
            ).scalar():
        SIGHTING_CACHE.add(pokemon)
        return
    obj = Sighting(
        pokemon_id=pokemon['pokemon_id'],
        spawn_id=pokemon['spawn_id'],
        encounter_id=pokemon['encounter_id'],
        expire_timestamp=pokemon['expire_timestamp'],
        lat=pokemon['lat'],
        lon=pokemon['lon'],
        atk_iv=pokemon.get('individual_attack'),
        def_iv=pokemon.get('individual_defense'),
        sta_iv=pokemon.get('individual_stamina'),
        move_1=pokemon.get('move_1'),
        move_2=pokemon.get('move_2')
    )
    session.add(obj)
    SIGHTING_CACHE.add(pokemon)

def add_mystery_spawnpoint(session, pokemon):
    # Check if the same entry already exists
    spawn_id = pokemon['spawn_id']
    point = pokemon['lat'], pokemon['lon']
    if point in spawns.unknown or session.query(exists().where(
            Spawnpoint.spawn_id == spawn_id)).scalar():
        return

    session.add(Spawnpoint(
        spawn_id=spawn_id,
        despawn_time=None,
        lat=pokemon['lat'],
        lon=pokemon['lon'],
        updated=0,
        duration=None,
        failures=0
    ))

    if point in bounds:
        spawns.add_unknown(point)

def add_fort_sighting(session, raw_fort):
    # Check if fort exists
    fort = session.query(Fort) \
        .filter(Fort.external_id == raw_fort['external_id']) \
        .first()
    if not fort:
        fort = Fort(
            external_id=raw_fort['external_id'],
            lat=raw_fort['lat'],
            lon=raw_fort['lon'],
        )
        session.add(fort)
    if fort.id and session.query(exists().where(and_(
                FortSighting.fort_id == fort.id,
                FortSighting.last_modified == raw_fort['last_modified']
            ))).scalar():
        # Why is it not in the cache? It should be there!
        FORT_CACHE.add(raw_fort)
        return
    obj = FortSighting(
        fort=fort,
        team=raw_fort['team'],
        prestige=raw_fort['prestige'],
        guard_pokemon_id=raw_fort['guard_pokemon_id'],
        last_modified=raw_fort['last_modified'],
    )
    session.add(obj)
    FORT_CACHE.add(raw_fort)

def add_pokestop(session, raw_pokestop):
    pokestop_id = raw_pokestop['external_id']
    if session.query(exists().where(
            Pokestop.external_id == pokestop_id)).scalar():
        FORT_CACHE.pokestops.add(pokestop_id)
        return

    pokestop = Pokestop(
        external_id=pokestop_id,
        lat=raw_pokestop['lat'],
        lon=raw_pokestop['lon']
    )
    session.add(pokestop)
    FORT_CACHE.pokestops.add(pokestop_id)

def _last_error_free_job(cls, harvest_job):
        # TODO weed out cancelled jobs somehow.
        # look for jobs with no gather errors
        jobs = \
            model.Session.query(HarvestJob) \
                 .filter(HarvestJob.source == harvest_job.source) \
                 .filter(HarvestJob.gather_started != None) \
                 .filter(HarvestJob.status == 'Finished') \
                 .filter(HarvestJob.id != harvest_job.id) \
                 .filter(
                     ~exists().where(
                         HarvestGatherError.harvest_job_id == HarvestJob.id)) \
                 .order_by(HarvestJob.gather_started.desc())
        # now check them until we find one with no fetch/import errors
        # (looping rather than doing sql, in case there are lots of objects
        # and lots of jobs)
        for job in jobs:
            for obj in job.objects:
                if obj.current is False and \
                        obj.report_status != 'not modified':
                    # unsuccessful, so go onto the next job
                    break
            else:
                return job

def set_local_file_availability_from_disk(checksums=None):
    bridge = Bridge(app_name=CONTENT_APP_NAME)

    LocalFileClass = bridge.get_class(LocalFile)

    if checksums is None:
        logging.info('Setting availability of LocalFile objects based on disk availability')
        files = bridge.session.query(LocalFileClass).all()
    elif type(checksums) == list:
        logging.info('Setting availability of {number} LocalFile objects based on disk availability'.format(number=len(checksums)))
        files = bridge.session.query(LocalFileClass).filter(LocalFileClass.id.in_(checksums)).all()
    else:
        logging.info('Setting availability of LocalFile object with checksum {checksum} based on disk availability'.format(checksum=checksums))
        files = [bridge.session.query(LocalFileClass).get(checksums)]

    checksums_to_update = [
        file.id for file in files if os.path.exists(get_content_storage_file_path(get_content_file_name(file)))
    ]

    bridge.end()

    mark_local_files_as_available(checksums_to_update)

def register():
    form = PlayerForm(csrf_enabled=False)

    if request.method == 'POST' and form.validate_on_submit():
        if db.session.query(exists().where(Player.alias == form.alias.data)).scalar():
            flash('Alias already taken! Are you registered already?', category='warn')
        else:
            record = Player(alias=form.alias.data.lower(), first_name=form.first_name.data,
                            last_name=form.last_name.data)
            db.session.add(record)
            db.session.commit()
            return redirect('/record_match')

        return render_template('register.html')

    else:
        flash_errors(form)

    return render_template('register.html')

def create_db(database_url):
    """Create an empty database with the tables defined above."""
    engine = create_engine(database_url)
    Base.metadata.create_all(engine)

    # Create a Null Clan to store members that not in a tracked.
    from sqlalchemy import exists
    from cocman.connect_to_database import connect_to_database
    session = connect_to_database()
    (already_exists, ), = session.query(exists().where(Clan.tag == '#NULL'))

    if already_exists is False:
        null_clan = Clan(name='Null Clan', tag='#NULL')
        session.add(null_clan)
        session.commit()

    session.close()

def add_user(id_, paraphrase, session):
    """Add new users to the database."""
    if session.query(exists().where(User.id == id_)).scalar():
        raise UserExists(id_=id_)
    else:
        new_user = User(id=id_, paraphrase=paraphrase)
        session.add(new_user)
        session.commit()

# TODO: Implement handhasking for better security
# def create_nonce(id_, session):
#     """Assign a random nonce to the user."""
#     user = None
#     try:
#         user = session.query(User).filter(User.id == id_).one()
#     except NoResultFound:
#         raise UserNotFound(id_=id_)
#     user.nonce = random.randint(1, 1000000)
#     session.commit()
#
#     return user.nonce

def tag_update_insert(userid, submitid):
    we = d.meta.tables['welcome']
    db = d.connect()
    q = sa.select([sa.exists(
        sa.select([1])
        .where(we.c.userid == userid)
        .where(we.c.otherid == submitid)
        .where(we.c.type == 3140))])
    if db.scalar(q):
        return
    db.execute(
        we.insert()
        .values(userid=userid, otherid=submitid, unixtime=arrow.utcnow(), type=3140))


# notifications
#   3150 site update

def exists(self, name):
        """
        Given a flags name, check if it exists in the store.
        """

        query = sa.select([sa.exists().where(self.flags.c.name == name)])
        res = self.connection.execute(query).fetchone()
        return res[0]

def has_name_version(self, name, version):
        """Checks if the name/version combination is already in the database

        :param str name: The network name
        :param str version: The network version
        :rtype: bool
        """
        return self.session.query(exists().where(and_(Network.name == name, Network.version == version))).scalar()

def orphaned_pages(self):
        """Gives all pages with no links to them."""

        stmt = ~exists().where(Link.target==Title.title)
        orphaned = self.db.query(Title.title).\
                filter(stmt).\
                order_by(Title.title)
        for (title,) in orphaned:
            yield unicode(title)

def wanted_pages(self):
        """Gives all pages that are linked to, but don't exist, together with
        the number of links."""

        stmt = ~exists().where(Title.title==Link.target)
        wanted = self.db.query(func.count(), Link.target).\
                filter(stmt).\
                group_by(Link.target).\
                order_by(-func.count())
        for refs, title, in wanted:
            title = unicode(title)
            if not external_link(title) and not title.startswith('+'):
                yield refs, title

def update_channel_metadata():
    """
    If we are potentially moving from a version of Kolibri that did not import its content data,
    scan through the settings.CONTENT_DATABASE_DIR folder for all channel content databases,
    and pull the data from each database if we have not already imported it.
    """
    from .channel_import import import_channel_from_local_db
    channel_ids = get_channel_ids_for_content_database_dir(settings.CONTENT_DATABASE_DIR)
    for channel_id in channel_ids:
        if not ChannelMetadata.objects.filter(id=channel_id).exists():
            import_channel_from_local_db(channel_id)
            set_availability(channel_id)

def set_leaf_node_availability_from_local_file_availability():
    bridge = Bridge(app_name=CONTENT_APP_NAME)

    ContentNodeTable = bridge.get_table(ContentNode)
    FileTable = bridge.get_table(File)
    LocalFileTable = bridge.get_table(LocalFile)

    connection = bridge.get_connection()

    file_statement = select([LocalFileTable.c.available]).where(
        FileTable.c.local_file_id == LocalFileTable.c.id,
    ).limit(1)

    logging.info('Setting availability of File objects based on LocalFile availability')

    connection.execute(FileTable.update().values(available=file_statement).execution_options(autocommit=True))

    contentnode_statement = select([FileTable.c.contentnode_id]).where(
        and_(
            FileTable.c.available == True,  # noqa
            FileTable.c.supplementary == False
        )
    ).where(ContentNodeTable.c.id == FileTable.c.contentnode_id)

    logging.info('Setting availability of non-topic ContentNode objects based on File availability')

    connection.execute(ContentNodeTable.update().where(
        ContentNodeTable.c.kind != content_kinds.TOPIC).values(available=exists(contentnode_statement)).execution_options(autocommit=True))

    bridge.end()

def _create(self, emails):
        with self._dbwrite() as db:
            for email in emails:
                uid_exists = exists().where(_Email.uid == email['_uid'])
                if not db.query(uid_exists).scalar():
                    db.add(_Email.from_dict(db, email))

def __exists_replicas(rse_id, scope=None, name=None, path=None, session=None):
    """
    Internal method to check if a replica exists at a given site.
    :param rse_id: The RSE id.
    :param scope: The scope of the file.
    :param name: The name of the file.
    :param path: The path of the replica.
    :param session: The database session in use.
    """

    already_declared = False
    if path:
        path_clause = [models.RSEFileAssociation.path == path]
        if path.startswith('/'):
            path_clause.append(models.RSEFileAssociation.path == path[1:])
        else:
            path_clause.append(models.RSEFileAssociation.path == '/%s' % path)
        query = session.query(models.RSEFileAssociation.path, models.RSEFileAssociation.scope, models.RSEFileAssociation.name, models.RSEFileAssociation.rse_id).\
            with_hint(models.RSEFileAssociation, "+ index(replicas REPLICAS_PATH_IDX", 'oracle').\
            filter(models.RSEFileAssociation.rse_id == rse_id).filter(or_(*path_clause))
    else:
        query = session.query(models.RSEFileAssociation.path, models.RSEFileAssociation.scope, models.RSEFileAssociation.name, models.RSEFileAssociation.rse_id).\
            filter_by(rse_id=rse_id, scope=scope, name=name)
    if query.count():
        result = query.first()
        path, scope, name, rse_id = result[0], result[1], result[2], result[3]
        # Now we check that the replica is not already declared bad
        query = session.query(models.BadReplicas.scope, models.BadReplicas.name, models.BadReplicas.rse_id, models.BadReplicas.state).\
            filter_by(rse_id=rse_id, scope=scope, name=name, state=BadFilesStatus.BAD)
        if query.count():
            already_declared = True
        return True, scope, name, already_declared
    else:
        return False, None, None, already_declared

def __bulk_add_new_file_dids(files, account, dataset_meta=None, session=None):
    """
    Bulk add new dids.

    :param dids: the list of new files.
    :param account: The account owner.
    :param session: The database session in use.
    :returns: True is successful.
    """
    for file in files:
        new_did = models.DataIdentifier(scope=file['scope'], name=file['name'],
                                        account=file.get('account') or account,
                                        did_type=DIDType.FILE, bytes=file['bytes'],
                                        md5=file.get('md5'), adler32=file.get('adler32'),
                                        is_new=None)
        for key in file.get('meta', []):
            new_did.update({key: file['meta'][key]})
        for key in dataset_meta or {}:
            new_did.update({key: dataset_meta[key]})

        new_did.save(session=session, flush=False)
    try:
        session.flush()
    except IntegrityError, error:
        raise exception.RucioException(error.args)
    except DatabaseError, error:
        raise exception.RucioException(error.args)
    except FlushError, error:
        if match('New instance .* with identity key .* conflicts with persistent instance', error.args[0]):
            raise exception.DataIdentifierAlreadyExists('Data Identifier already exists!')
        raise exception.RucioException(error.args)
    return True

def current(cls, group=None):
        q = exists()
        q = q.select_from(join(GroupPatient, Group, GroupPatient.group_id == Group.id))
        q = q.where(GroupPatient.patient_id == cls.id)
        q = q.where(GroupPatient.current == True)  # noqa

        if group is not None:
            q = q.where(Group.id == group.id)
        else:
            q = q.where(Group.type == GROUP_TYPE.SYSTEM)

        return q

def insert_classes(classes, session):
    """Insert all the classes as defined in the APIDocumentation into DB."""
    # print(session.query(exists().where(RDFClass.name == "Datastream")).scalar())
    class_list = [RDFClass(name=class_["label"].strip('.')) for class_ in classes
                  if "label" in class_ and
                  not session.query(exists().where(RDFClass.name == class_["label"].strip('.'))).scalar()]

    class_list = class_list + [RDFClass(name=class_["title"].strip('.')) for class_ in classes
                               if "title" in class_ and
                               not session.query(exists().where(RDFClass.name == class_["title"].strip('.'))).scalar()]
    # print(class_list)
    session.add_all(class_list)
    session.commit()
    return None

def insert_properties(properties, session):
    """Insert all the properties as defined in the APIDocumentation into DB."""
    prop_list = [BaseProperty(name=prop) for prop in properties
                 if not session.query(exists().where(BaseProperty.name == prop)).scalar()]
    session.add_all(prop_list)
    session.commit()
    return None


# if __name__ == "__main__":
#     Session = sessionmaker(bind=engine)
#     session = Session()
#
#     doc = doc_gen("test", "test")
#     # Extract all classes with supportedProperty from both
#     classes = get_classes(doc.generate())
#
#     # Extract all properties from both
#     # import pdb; pdb.set_trace()
#     properties = get_all_properties(classes)
#     # Add all the classes
#     insert_classes(classes, session)
#     print("Classes inserted successfully")
#     # Add all the properties
#     insert_properties(properties, session)
#     print("Properties inserted successfully")

def exists(self, post):
        return self.session.query(exists().where(self.PostORM.author==post.author)
                .where(self.PostORM.title==post.title)
                .where(self.PostORM.subreddit==post.subreddit)).scalar()

def dump_exists(cls, session, cluster_name):
        logger.info("Checking for MySQL dump for cluster {c}".format(
            c=cluster_name
        ))
        with session.connect_begin(ro=True) as s:
            mysql_dump_exists = s.query(
                exists().where(
                    MySQLDumps.cluster_name == cluster_name
                )
            ).scalar()
            logger.info("MySQL dump exists") if mysql_dump_exists else \
                logger.info("MySQL dump doesn't exist")
        return mysql_dump_exists

def delete_not_exists(table, selectable):
    """Statement to delete rows in table that are not in query result."""

    delete_statement = table.delete().where(
        ~exists(
            select(
                [literal_column('1')]
            ).select_from(primary_key_join(table, selectable))
        )
    )

    return delete_statement

def add_spawnpoint(session, pokemon):
    # Check if the same entry already exists
    spawn_id = pokemon['spawn_id']
    new_time = pokemon['expire_timestamp'] % 3600
    try:
        if new_time == spawns.despawn_times[spawn_id]:
            return
    except KeyError:
        pass
    existing = session.query(Spawnpoint) \
        .filter(Spawnpoint.spawn_id == spawn_id) \
        .first()
    now = round(time())
    point = pokemon['lat'], pokemon['lon']
    spawns.add_known(spawn_id, new_time, point)
    if existing:
        existing.updated = now
        existing.failures = 0

        if (existing.despawn_time is None or
                existing.updated < conf.LAST_MIGRATION):
            widest = get_widest_range(session, spawn_id)
            if widest and widest > 1800:
                existing.duration = 60
        elif new_time == existing.despawn_time:
            return

        existing.despawn_time = new_time
    else:
        widest = get_widest_range(session, spawn_id)

        duration = 60 if widest and widest > 1800 else None

        session.add(Spawnpoint(
            spawn_id=spawn_id,
            despawn_time=new_time,
            lat=pokemon['lat'],
            lon=pokemon['lon'],
            updated=now,
            duration=duration,
            failures=0
        ))

def recurse_availability_up_tree(channel_id):
    bridge = Bridge(app_name=CONTENT_APP_NAME)

    ContentNodeClass = bridge.get_class(ContentNode)

    ContentNodeTable = bridge.get_table(ContentNode)

    connection = bridge.get_connection()

    node_depth = bridge.session.query(func.max(ContentNodeClass.level)).scalar()

    logging.info('Setting availability of ContentNode objects with children for {levels} levels'.format(levels=node_depth))

    child = ContentNodeTable.alias()

    # start a transaction

    trans = connection.begin()
    # Go from the deepest level to the shallowest
    start = datetime.datetime.now()
    for level in range(node_depth, 0, -1):

        available_nodes = select([child.c.available]).where(
            and_(
                child.c.available == True,  # noqa
                child.c.level == level,
                child.c.channel_id == channel_id,
            )
        ).where(ContentNodeTable.c.id == child.c.parent_id)

        logging.info('Setting availability of ContentNode objects with children for level {level}'.format(level=level))
        # Only modify topic availability here
        connection.execute(ContentNodeTable.update().where(
            and_(
                ContentNodeTable.c.level == level - 1,
                ContentNodeTable.c.channel_id == channel_id,
                ContentNodeTable.c.kind == content_kinds.TOPIC)).values(available=exists(available_nodes)))

    # commit the transaction
    trans.commit()

    elapsed = (datetime.datetime.now() - start)
    logging.debug("Availability annotation took {} seconds".format(elapsed.seconds))

    bridge.end()

def __bulk_add_replicas(rse_id, files, account, session=None):
    """
    Bulk add new dids.

    :param rse_id: the RSE id.
    :param dids: the list of files.
    :param account: The account owner.
    :param session: The database session in use.
    :returns: True is successful.
    """
    nbfiles, bytes = 0, 0
    # Check for the replicas already available
    condition = or_()
    for f in files:
        condition.append(and_(models.RSEFileAssociation.scope == f['scope'], models.RSEFileAssociation.name == f['name'], models.RSEFileAssociation.rse_id == rse_id))

    query = session.query(models.RSEFileAssociation.scope, models.RSEFileAssociation.name, models.RSEFileAssociation.rse_id).\
        with_hint(models.RSEFileAssociation, text="INDEX(REPLICAS REPLICAS_PK)", dialect_name='oracle').\
        filter(condition)
    available_replicas = [dict([(column, getattr(row, column)) for column in row._fields]) for row in query]

    new_replicas = []
    for file in files:
        found = False
        for available_replica in available_replicas:
            if file['scope'] == available_replica['scope'] and file['name'] == available_replica['name'] and rse_id == available_replica['rse_id']:
                found = True
                break
        if not found:
            nbfiles += 1
            bytes += file['bytes']
            new_replicas.append({'rse_id': rse_id, 'scope': file['scope'],
                                 'name': file['name'], 'bytes': file['bytes'],
                                 'path': file.get('path'),
                                 'state': ReplicaState.from_string(file.get('state', 'A')),
                                 'md5': file.get('md5'), 'adler32': file.get('adler32'),
                                 'lock_cnt': file.get('lock_cnt', 0),
                                 'tombstone': file.get('tombstone')})
#            new_replica = models.RSEFileAssociation(rse_id=rse_id, scope=file['scope'], name=file['name'], bytes=file['bytes'],
#                                                    path=file.get('path'), state=ReplicaState.from_string(file.get('state', 'A')),
#                                                    md5=file.get('md5'), adler32=file.get('adler32'), lock_cnt=file.get('lock_cnt', 0),
#                                                    tombstone=file.get('tombstone'))
#            new_replica.save(session=session, flush=False)
    try:
        new_replicas and session.bulk_insert_mappings(models.RSEFileAssociation,
                                                      new_replicas)
        session.flush()
        return nbfiles, bytes
    except IntegrityError, error:
        if match('.*IntegrityError.*ORA-00001: unique constraint .*REPLICAS_PK.*violated.*', error.args[0]) \
           or match('.*IntegrityError.*1062.*Duplicate entry.*', error.args[0]) \
           or error.args[0] == '(IntegrityError) columns rse_id, scope, name are not unique' \
           or match('.*IntegrityError.*duplicate key value violates unique constraint.*', error.args[0]):
            raise exception.Duplicate("File replica already exists!")
        raise exception.RucioException(error.args)
    except DatabaseError, error:
        raise exception.RucioException(error.args)

def list_unlocked_replicas(rse, limit, bytes=None, rse_id=None, worker_number=None, total_workers=None, delay_seconds=0, session=None):
    """
    List RSE File replicas with no locks.

    :param rse: the rse name.
    :param bytes: the amount of needed bytes.
    :param session: The database session in use.

    :returns: a list of dictionary replica.
    """
    if not rse_id:
        rse_id = get_rse_id(rse=rse, session=session)

    # filter(models.RSEFileAssociation.state != ReplicaState.BEING_DELETED).\
    none_value = None  # Hack to get pep8 happy...
    query = session.query(models.RSEFileAssociation.scope, models.RSEFileAssociation.name, models.RSEFileAssociation.path, models.RSEFileAssociation.bytes, models.RSEFileAssociation.tombstone, models.RSEFileAssociation.state).\
        with_hint(models.RSEFileAssociation, "INDEX_RS_ASC(replicas REPLICAS_TOMBSTONE_IDX)  NO_INDEX_FFS(replicas REPLICAS_TOMBSTONE_IDX)", 'oracle').\
        filter(models.RSEFileAssociation.tombstone < datetime.utcnow()).\
        filter(models.RSEFileAssociation.lock_cnt == 0).\
        filter(case([(models.RSEFileAssociation.tombstone != none_value, models.RSEFileAssociation.rse_id), ]) == rse_id).\
        filter(or_(models.RSEFileAssociation.state.in_((ReplicaState.AVAILABLE, ReplicaState.UNAVAILABLE, ReplicaState.BAD)),
                   and_(models.RSEFileAssociation.state == ReplicaState.BEING_DELETED, models.RSEFileAssociation.updated_at < datetime.utcnow() - timedelta(seconds=delay_seconds)))).\
        order_by(models.RSEFileAssociation.tombstone)

    # do no delete files used as sources
    stmt = exists(select([1]).prefix_with("/*+ INDEX(requests REQUESTS_SCOPE_NAME_RSE_IDX) */", dialect='oracle')).\
        where(and_(models.RSEFileAssociation.scope == models.Request.scope,
                   models.RSEFileAssociation.name == models.Request.name))
    query = query.filter(not_(stmt))

    if worker_number and total_workers and total_workers - 1 > 0:
        if session.bind.dialect.name == 'oracle':
            bindparams = [bindparam('worker_number', worker_number - 1), bindparam('total_workers', total_workers - 1)]
            query = query.filter(text('ORA_HASH(name, :total_workers) = :worker_number', bindparams=bindparams))
        elif session.bind.dialect.name == 'mysql':
            query = query.filter(text('mod(md5(name), %s) = %s' % (total_workers - 1, worker_number - 1)))
        elif session.bind.dialect.name == 'postgresql':
            query = query.filter(text('mod(abs((\'x\'||md5(name))::bit(32)::int), %s) = %s' % (total_workers - 1, worker_number - 1)))

    needed_space = bytes
    total_bytes, total_files = 0, 0
    rows = []
    for (scope, name, path, bytes, tombstone, state) in query.yield_per(1000):
        if state != ReplicaState.UNAVAILABLE:

            total_bytes += bytes
            if tombstone != OBSOLETE and needed_space is not None and total_bytes > needed_space:
                break

            total_files += 1
            if total_files > limit:
                break

        rows.append({'scope': scope, 'name': name, 'path': path,
                     'bytes': bytes, 'tombstone': tombstone,
                     'state': state})
    return rows

def update_replicas_states(replicas, nowait=False, session=None):
    """
    Update File replica information and state.

    :param replicas: The list of replicas.
    :param nowait:   Nowait parameter for the for_update queries.
    :param session:  The database session in use.
    """
    rse_ids = {}
    for replica in replicas:
        if 'rse_id' not in replica:
            if replica['rse'] not in rse_ids:
                rse_ids[replica['rse']] = get_rse_id(rse=replica['rse'], session=session)
            replica['rse_id'] = rse_ids[replica['rse']]

        query = session.query(models.RSEFileAssociation).filter_by(rse_id=replica['rse_id'], scope=replica['scope'], name=replica['name'])
        try:
            if nowait:
                query.with_for_update(nowait=True).one()
        except NoResultFound:
            # remember scope, name and rse_id
            raise exception.ReplicaNotFound("No row found for scope: %s name: %s rse_id: %s" % (replica['scope'], replica['name'], replica['rse_id']))

        if isinstance(replica['state'], str) or isinstance(replica['state'], unicode):
            replica['state'] = ReplicaState.from_string(replica['state'])

        values = {'state': replica['state']}
        if replica['state'] == ReplicaState.BEING_DELETED:
            query = query.filter_by(lock_cnt=0)
            # Exclude replicas use as sources
            stmt = exists([1]).where(and_(models.RSEFileAssociation.scope == models.Source.scope,
                                          models.RSEFileAssociation.name == models.Source.name,
                                          models.RSEFileAssociation.rse_id == models.Source.rse_id))
            query = query.filter(not_(stmt))
            values['tombstone'] = OBSOLETE
        elif replica['state'] == ReplicaState.AVAILABLE:
            rucio.core.lock.successful_transfer(scope=replica['scope'], name=replica['name'], rse_id=replica['rse_id'], nowait=nowait, session=session)
        elif replica['state'] == ReplicaState.UNAVAILABLE:
            rucio.core.lock.failed_transfer(scope=replica['scope'], name=replica['name'], rse_id=replica['rse_id'],
                                            error_message=replica.get('error_message', None),
                                            broken_rule_id=replica.get('broken_rule_id', None),
                                            broken_message=replica.get('broken_message', None),
                                            nowait=nowait, session=session)

        if 'path' in replica and replica['path']:
            values['path'] = replica['path']

        if not query.update(values, synchronize_session=False):
            if 'rse' not in replica:
                replica['rse'] = get_rse_name(rse_id=replica['rse_id'], session=session)
            raise exception.UnsupportedOperation('State %(state)s for replica %(scope)s:%(name)s on %(rse)s cannot be updated' % replica)
    return True

def add_volatile_replicas(rse, replicas, session=None):
    """
    Bulk add volatile replicas.

    :param rse: the rse name.
    :param replicas: the list of volatile replicas.
    :param session: The database session in use.
    :returns: True is successful.
    """
    # first check that the rse is a volatile one
    try:
        rse_id = session.query(models.RSE.id).filter_by(rse=rse, volatile=True).one()[0]
    except NoResultFound:
        raise exception.UnsupportedOperation('No volatile rse found for %(rse)s !' % locals())

    file_clause, replica_clause = [], []
    for replica in replicas:
        file_clause.append(and_(models.DataIdentifier.scope == replica['scope'],
                                models.DataIdentifier.name == replica['name'],
                                ~exists(select([1]).prefix_with("/*+ INDEX(REPLICAS REPLICAS_PK) */", dialect='oracle')).where(and_(models.RSEFileAssociation.scope == replica['scope'],
                                                                                                                                    models.RSEFileAssociation.name == replica['name'],
                                                                                                                                    models.RSEFileAssociation.rse_id == rse_id))))
        replica_clause.append(and_(models.RSEFileAssociation.scope == replica['scope'],
                                   models.RSEFileAssociation.name == replica['name'],
                                   models.RSEFileAssociation.rse_id == rse_id))

    if replica_clause:
        now = datetime.utcnow()
        session.query(models.RSEFileAssociation).\
            with_hint(models.RSEFileAssociation, "index(REPLICAS REPLICAS_PK)", 'oracle').\
            filter(or_(*replica_clause)).\
            update({'updated_at': now, 'tombstone': now}, synchronize_session=False)

    if file_clause:
        file_query = session.query(models.DataIdentifier.scope,
                                   models.DataIdentifier.name,
                                   models.DataIdentifier.bytes,
                                   models.DataIdentifier.md5,
                                   models.DataIdentifier.adler32).\
            filter(or_(*file_clause))

        session.bulk_insert_mappings(
            models.RSEFileAssociation,
            [{'rse_id': rse_id, 'adler32': adler32, 'state': ReplicaState.AVAILABLE,
              'scope': scope, 'name': name, 'lock_cnt': 0, 'tombstone': datetime.utcnow(),
              'bytes': bytes, 'md5': md5} for scope, name, bytes, md5, adler32 in file_query])

def list_expired_dids(worker_number=None, total_workers=None, limit=None, session=None):
    """
    List expired data identifiers.

    :param limit: limit number.
    :param session: The database session in use.
    """

    stmt = exists().where(and_(models.ReplicationRule.scope == models.DataIdentifier.scope,
                               models.ReplicationRule.name == models.DataIdentifier.name,
                               models.ReplicationRule.locked == true()))
    query = session.query(models.DataIdentifier.scope, models.DataIdentifier.name,
                          models.DataIdentifier.did_type,
                          models.DataIdentifier.created_at,
                          models.DataIdentifier.purge_replicas).\
        filter(models.DataIdentifier.expired_at < datetime.utcnow(), not_(stmt)).\
        order_by(models.DataIdentifier.expired_at).\
        with_hint(models.DataIdentifier, "index(DIDS DIDS_EXPIRED_AT_IDX)", 'oracle')

    if worker_number and total_workers and total_workers - 1 > 0:
        if session.bind.dialect.name == 'oracle':
            bindparams = [bindparam('worker_number', worker_number - 1), bindparam('total_workers', total_workers - 1)]
            query = query.filter(text('ORA_HASH(name, :total_workers) = :worker_number', bindparams=bindparams))
        elif session.bind.dialect.name == 'mysql':
            query = query.filter(text('mod(md5(name), %s) = %s' % (total_workers - 1, worker_number - 1)))
        elif session.bind.dialect.name == 'postgresql':
            query = query.filter(text('mod(abs((\'x\'||md5(name))::bit(32)::int), %s) = %s' % (total_workers - 1, worker_number - 1)))
        elif session.bind.dialect.name == 'sqlite':
            row_count = 0
            dids = list()
            for scope, name, did_type, created_at, purge_replicas in query.yield_per(10):
                if int(md5(name).hexdigest(), 16) % total_workers == worker_number - 1:
                    dids.append({'scope': scope,
                                 'name': name,
                                 'did_type': did_type,
                                 'created_at': created_at,
                                 'purge_replicas': purge_replicas})
                    row_count += 1
                if limit and row_count >= limit:
                    return dids
            return dids

    if limit:
        query = query.limit(limit)

    return [{'scope': scope, 'name': name, 'did_type': did_type, 'created_at': created_at,
             'purge_replicas': purge_replicas} for scope, name, did_type, created_at, purge_replicas in query]

def list_new_dids(did_type, thread=None, total_threads=None, chunk_size=1000, session=None):
    """
    List recent identifiers.

    :param did_type : The DID type.
    :param thread: The assigned thread for this necromancer.
    :param total_threads: The total number of threads of all necromancers.
    :param chunk_size: Number of requests to return per yield.
    :param session: The database session in use.
    """

    stmt = select([1]).\
        prefix_with("/*+ INDEX(RULES ATLAS_RUCIO.RULES_SCOPE_NAME_IDX) */",
                    dialect='oracle').\
        where(and_(models.DataIdentifier.scope == models.ReplicationRule.scope,
                   models.DataIdentifier.name == models.ReplicationRule.name,
                   models.ReplicationRule.state == RuleState.INJECT))

    query = session.query(models.DataIdentifier).\
        with_hint(models.DataIdentifier, "index(dids DIDS_IS_NEW_IDX)", 'oracle').\
        filter_by(is_new=True).\
        filter(~exists(stmt))

    if did_type:
        if isinstance(did_type, str) or isinstance(did_type, unicode):
            query = query.filter_by(did_type=DIDType.from_sym(did_type))
        elif isinstance(did_type, EnumSymbol):
            query = query.filter_by(did_type=did_type)

    if total_threads and (total_threads - 1) > 0:
        if session.bind.dialect.name == 'oracle':
            bindparams = [bindparam('thread_number', thread), bindparam('total_threads', total_threads - 1)]
            query = query.filter(text('ORA_HASH(name, :total_threads) = :thread_number', bindparams=bindparams))
        elif session.bind.dialect.name == 'mysql':
            query = query.filter(text('mod(md5(name), %s) = %s' % (total_threads - 1, thread)))
        elif session.bind.dialect.name == 'postgresql':
            query = query.filter(text('mod(abs((\'x\'||md5(name))::bit(32)::int), %s) = %s' % (total_threads - 1, thread)))

    row_count = 0
    for chunk in query.yield_per(10):
        row_count += 1
        if row_count <= chunk_size:
            yield {'scope': chunk.scope, 'name': chunk.name, 'did_type': chunk.did_type}  # TODO Change this to the proper filebytes [RUCIO-199]
        else:
            break

def create_or_update(context, artifact_id, values, session):
    with session.begin():
        _drop_protected_attrs(models.Artifact, values)
        if artifact_id is None:
            # create new artifact
            artifact = models.Artifact()
            artifact.id = values.pop('id')
        else:
            # update the existing artifact
            artifact = _get(context, artifact_id, session)

        if 'version' in values:
            values['version'] = semver_db.parse(values['version'])

        if 'tags' in values:
            tags = values.pop('tags')
            artifact.tags = _do_tags(artifact, tags)

        if 'properties' in values:
            properties = values.pop('properties', {})
            artifact.properties = _do_properties(artifact, properties)

        if 'blobs' in values:
            blobs = values.pop('blobs')
            artifact.blobs = _do_blobs(artifact, blobs)

        artifact.updated_at = timeutils.utcnow()
        if 'status' in values:
            if session.query(exists().where(and_(
                models.ArtifactBlob.status == 'saving',
                models.ArtifactBlob.artifact_id == artifact_id))
            ).one()[0]:
                raise exception.Conflict(
                    "You cannot change artifact status if it has "
                    "uploading blobs.")
            if values['status'] == 'active':
                artifact.activated_at = timeutils.utcnow()
        artifact.update(values)

        artifact.save(session=session)
        LOG.debug("Response from the database was received.")

        return artifact.to_dict()

def __init__(self, database_dir='./output'):
        if not os.path.exists(database_dir):
            os.makedirs(database_dir)

        path = os.path.join(database_dir, DATABASE_NAME)

        self.engine = sqlalchemy.create_engine('sqlite:///'+path, module=sqlite, echo=False)
        self.Sessionmaker = sessionmaker(bind=self.engine)

        Base = declarative_base()
        class PostORM(Base):
            __tablename__ = 'posts'
            __table_args__ = {'sqlite_autoincrement': True}

            id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
            name = sqlalchemy.Column(sqlalchemy.String)
            password = sqlalchemy.Column(sqlalchemy.String)
            title = sqlalchemy.Column(sqlalchemy.String)
            author = sqlalchemy.Column(sqlalchemy.String)
            url = sqlalchemy.Column(sqlalchemy.String)
            shortlink = sqlalchemy.Column(sqlalchemy.String)
            subreddit = sqlalchemy.Column(sqlalchemy.String)
            folder = sqlalchemy.Column(sqlalchemy.String)
            num_files = sqlalchemy.Column(sqlalchemy.String)
            date_added = sqlalchemy.Column(DATETIME)

            def __init__(self, title, author, url, shortlink, subreddit, num_files, folder):
                self.title = title
                self.author = author
                self.url = url
                self.shortlink = shortlink
                self.subreddit = subreddit
                self.folder = folder
                self.num_files = num_files
                self.date_added = datetime.now()

        self.PostORM = PostORM

        self.Base = Base
        if not self.engine.dialect.has_table(self.engine, PostORM.__tablename__):
            self.Base.metadata.create_all(self.engine)
        self.session = self.Sessionmaker()