<commit_before>"""
NOTE:

Avoid using this module on your own or in plugins, this was originally made for 0.9 -> 1.0 transition.

You can safely use task.simple_persistence and manager.persist, if we implement something better we
can replace underlying mechanism in single point (and provide transparent switch).
"""

from __future__ import unicode_literals, division, absolute_import
from collections import MutableMapping, defaultdict
from datetime import datetime
import logging
import pickle

from sqlalchemy import Column, Integer, String, DateTime, Unicode, select, Index

from flexget import db_schema
from flexget.event import event
from flexget.manager import Session
from flexget.utils import json
from flexget.utils.database import json_synonym
from flexget.utils.sqlalchemy_utils import table_schema, create_index, table_add_column

log = logging.getLogger('util.simple_persistence')
Base = db_schema.versioned_base('simple_persistence', 4)

# Used to signify that a given key should be deleted from simple persistence on flush
DELETE = object()


@db_schema.upgrade('simple_persistence')
def upgrade(ver, session):
    if ver is None:
        # Upgrade to version 0 was a failed attempt at cleaning bad entries from our table, better attempt in ver 1
        ver = 0
    if ver == 0:
        # Remove any values that are not loadable.
        table = table_schema('simple_persistence', session)
        for row in session.execute(select([table.c.id, table.c.plugin, table.c.key, table.c.value])):
            try:
                p = pickle.loads(row['value'])
            except Exception as e:
                log.warning('Couldn\'t load %s:%s removing from db: %s' % (row['plugin'], row['key'], e))
                session.execute(table.delete().where(table.c.id == row['id']))
        ver = 1
    if ver == 1:
        log.info('Creating index on simple_persistence table.')
        create_index('simple_persistence', session, 'feed', 'plugin', 'key')
        ver = 2
    if ver == 2 or ver == 3:
        table = table_schema('simple_persistence', session)
        table_add_column(table, 'json', Unicode, session)
        # Make sure we get the new schema with the added column
        table = table_schema('simple_persistence', session)
        for row in session.execute(select([table.c.id, table.c.value])):
            p = pickle.loads(row['value'])
            session.execute(table.update().where(table.c.id == row['id']).values(
                json=json.dumps(p, encode_datetime=True)))
        ver = 4
    return ver


@event('manager.db_cleanup')
def db_cleanup(manager, session):
    """Clean up values in the db from tasks which no longer exist."""
    # SKVs not associated with any task use None as task tame
    existing_tasks = list(manager.tasks) + [None]
    session.query(SimpleKeyValue).filter(~SimpleKeyValue.task.in_(existing_tasks)).delete(synchronize_session=False)


class SimpleKeyValue(Base):
    __tablename__ = 'simple_persistence'

    id = Column(Integer, primary_key=True)
    task = Column('feed', String)
    plugin = Column(String)
    key = Column(String)
    _json = Column('json', Unicode)
    value = json_synonym('_json')
    added = Column(DateTime, default=datetime.now())

    def __init__(self, task, plugin, key, value):
        self.task = task
        self.plugin = plugin
        self.key = key
        self.value = value

    def __repr__(self):
        return "<SimpleKeyValue('%s','%s','%s')>" % (self.task, self.key, self.value)

Index('ix_simple_persistence_feed_plugin_key', SimpleKeyValue.task, SimpleKeyValue.plugin, SimpleKeyValue.key)


class SimplePersistence(MutableMapping):
    """
    Store simple values that need to be persisted between FlexGet runs. Interface is like a `dict`.

    This should only be used if a plugin needs to store a few values, otherwise it should create a full table in
    the database.
    """
    # Stores values in store[taskname][pluginname][key] format
    class_store = defaultdict(lambda: defaultdict(dict))

    def __init__(self, plugin=None):
        self.taskname = None
        self.plugin = plugin

    @property
    def store(self):
        return self.class_store[self.taskname][self.plugin]

    def __setitem__(self, key, value):
        log.debug('setting key %s value %s' % (key, repr(value)))
        self.store[key] = value

    def __getitem__(self, key):
        if key not in self.store or self.store[key] == DELETE:
            raise KeyError('%s is not contained in the simple_persistence table.' % key)
        return self.store[key]

    def __delitem__(self, key):
        self.store[key] = DELETE

    def __iter__(self):
        return iter(self.store)

    def __len__(self):
        return len(self.store)

    @classmethod
    def load(cls, task=None):
        """Load all key/values from `task` into memory from database."""
        with Session() as session:
            for skv in session.query(SimpleKeyValue).filter(SimpleKeyValue.task == task).all():
                cls.class_store[task][skv.plugin][skv.key] = skv.value

    @classmethod
    def flush(cls, task=None):
        """Flush all in memory key/values to database."""
        log.debug('Flushing simple persistence for task %s to db.' % task)
        with Session() as session:
            for pluginname in cls.class_store[task]:
                for key, value in cls.class_store[task][pluginname].iteritems():
                    query = (session.query(SimpleKeyValue).
                        filter(SimpleKeyValue.task == task).
                        filter(SimpleKeyValue.plugin == pluginname).
                        filter(SimpleKeyValue.key == key))
                    if value == DELETE:
                        query.delete()
                    else:
                        updated = query.update({'value': unicode(json.dumps(value))}, synchronize_session=False)
                        if not updated:
                            session.add(SimpleKeyValue(task, pluginname, key, value))


class SimpleTaskPersistence(SimplePersistence):<commit_msg>support datetime in updates<commit_after>"""
NOTE:

Avoid using this module on your own or in plugins, this was originally made for 0.9 -> 1.0 transition.

You can safely use task.simple_persistence and manager.persist, if we implement something better we
can replace underlying mechanism in single point (and provide transparent switch).
"""

from __future__ import unicode_literals, division, absolute_import
from collections import MutableMapping, defaultdict
from datetime import datetime
import logging
import pickle

from sqlalchemy import Column, Integer, String, DateTime, Unicode, select, Index

from flexget import db_schema
from flexget.event import event
from flexget.manager import Session
from flexget.utils import json
from flexget.utils.database import json_synonym
from flexget.utils.sqlalchemy_utils import table_schema, create_index, table_add_column

log = logging.getLogger('util.simple_persistence')
Base = db_schema.versioned_base('simple_persistence', 4)

# Used to signify that a given key should be deleted from simple persistence on flush
DELETE = object()


@db_schema.upgrade('simple_persistence')
def upgrade(ver, session):
    if ver is None:
        # Upgrade to version 0 was a failed attempt at cleaning bad entries from our table, better attempt in ver 1
        ver = 0
    if ver == 0:
        # Remove any values that are not loadable.
        table = table_schema('simple_persistence', session)
        for row in session.execute(select([table.c.id, table.c.plugin, table.c.key, table.c.value])):
            try:
                p = pickle.loads(row['value'])
            except Exception as e:
                log.warning('Couldn\'t load %s:%s removing from db: %s' % (row['plugin'], row['key'], e))
                session.execute(table.delete().where(table.c.id == row['id']))
        ver = 1
    if ver == 1:
        log.info('Creating index on simple_persistence table.')
        create_index('simple_persistence', session, 'feed', 'plugin', 'key')
        ver = 2
    if ver == 2 or ver == 3:
        table = table_schema('simple_persistence', session)
        table_add_column(table, 'json', Unicode, session)
        # Make sure we get the new schema with the added column
        table = table_schema('simple_persistence', session)
        for row in session.execute(select([table.c.id, table.c.value])):
            p = pickle.loads(row['value'])
            session.execute(table.update().where(table.c.id == row['id']).values(
                json=json.dumps(p, encode_datetime=True)))
        ver = 4
    return ver


@event('manager.db_cleanup')
def db_cleanup(manager, session):
    """Clean up values in the db from tasks which no longer exist."""
    # SKVs not associated with any task use None as task tame
    existing_tasks = list(manager.tasks) + [None]
    session.query(SimpleKeyValue).filter(~SimpleKeyValue.task.in_(existing_tasks)).delete(synchronize_session=False)


class SimpleKeyValue(Base):
    __tablename__ = 'simple_persistence'

    id = Column(Integer, primary_key=True)
    task = Column('feed', String)
    plugin = Column(String)
    key = Column(String)
    _json = Column('json', Unicode)
    value = json_synonym('_json')
    added = Column(DateTime, default=datetime.now())

    def __init__(self, task, plugin, key, value):
        self.task = task
        self.plugin = plugin
        self.key = key
        self.value = value

    def __repr__(self):
        return "<SimpleKeyValue('%s','%s','%s')>" % (self.task, self.key, self.value)

Index('ix_simple_persistence_feed_plugin_key', SimpleKeyValue.task, SimpleKeyValue.plugin, SimpleKeyValue.key)


class SimplePersistence(MutableMapping):
    """
    Store simple values that need to be persisted between FlexGet runs. Interface is like a `dict`.

    This should only be used if a plugin needs to store a few values, otherwise it should create a full table in
    the database.
    """
    # Stores values in store[taskname][pluginname][key] format
    class_store = defaultdict(lambda: defaultdict(dict))

    def __init__(self, plugin=None):
        self.taskname = None
        self.plugin = plugin

    @property
    def store(self):
        return self.class_store[self.taskname][self.plugin]

    def __setitem__(self, key, value):
        log.debug('setting key %s value %s' % (key, repr(value)))
        self.store[key] = value

    def __getitem__(self, key):
        if key not in self.store or self.store[key] == DELETE:
            raise KeyError('%s is not contained in the simple_persistence table.' % key)
        return self.store[key]

    def __delitem__(self, key):
        self.store[key] = DELETE

    def __iter__(self):
        return iter(self.store)

    def __len__(self):
        return len(self.store)

    @classmethod
    def load(cls, task=None):
        """Load all key/values from `task` into memory from database."""
        with Session() as session:
            for skv in session.query(SimpleKeyValue).filter(SimpleKeyValue.task == task).all():
                cls.class_store[task][skv.plugin][skv.key] = skv.value

    @classmethod
    def flush(cls, task=None):
        """Flush all in memory key/values to database."""
        log.debug('Flushing simple persistence for task %s to db.' % task)
        with Session() as session:
            for pluginname in cls.class_store[task]:
                for key, value in cls.class_store[task][pluginname].iteritems():
                    query = (session.query(SimpleKeyValue).
                        filter(SimpleKeyValue.task == task).
                        filter(SimpleKeyValue.plugin == pluginname).
                        filter(SimpleKeyValue.key == key))
                    if value == DELETE:
                        query.delete()
                    else:
                        updated = query.update({'value': unicode(json.dumps(value), encode_datetime=True)}, synchronize_session=False)
                        if not updated:
                            session.add(SimpleKeyValue(task, pluginname, key, value))


class SimpleTaskPersistence(SimplePersistence):
{'fraction_non_alphanumeric': 0.1050730586110655, 'fraction_numerical': 0.002462649811196848, 'mean_word_length': 2.8713060057197333, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4650960', 'n_tokens_mistral': 3429, 'n_tokens_neox': 3328, 'n_words': 1045}