sopel.db.SopelDB.__init__() - Code Metrics - sopel-irc/sopel - Measure and Improve Code Quality continuously with Scrutinizer

sopel.db.SopelDB.init() F
last analyzed 2019-10-21 18:25 UTC

↳ Parent: sopel.db

Complexity

Conditions

Size

Total Lines	70
Code Lines	50

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	50
dl	0
loc	70
rs	2.9998
c	0
b	0
f	0
cc	15
nop	2

How to fix Long Method Complexity

# coding=utf-8
from __future__ import unicode_literals, absolute_import, print_function, division

import errno
import json
import os.path
import sys

from sopel.tools import Identifier

from sqlalchemy import create_engine, Column, ForeignKey, Integer, String
from sqlalchemy.engine.url import URL
from sqlalchemy.exc import OperationalError, SQLAlchemyError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker

if sys.version_info.major >= 3:
    unicode = str
    basestring = str


def _deserialize(value):
    if value is None:
        return None
    # sqlite likes to return ints for strings that look like ints, even though
    # the column type is string. That's how you do dynamic typing wrong.
    value = unicode(value)
    # Just in case someone's mucking with the DB in a way we can't account for,
    # ignore json parsing errors
    try:
        value = json.loads(value)
    except ValueError:
        pass
    return value


BASE = declarative_base()
MYSQL_TABLE_ARGS = {'mysql_engine': 'InnoDB',
                    'mysql_charset': 'utf8mb4',
                    'mysql_collate': 'utf8mb4_unicode_ci'}


class NickIDs(BASE):
    """
    NickIDs SQLAlchemy Class
    """
    __tablename__ = 'nick_ids'
    nick_id = Column(Integer, primary_key=True)


class Nicknames(BASE):
    """
    Nicknames SQLAlchemy Class
    """
    __tablename__ = 'nicknames'
    __table_args__ = MYSQL_TABLE_ARGS
    nick_id = Column(Integer, ForeignKey('nick_ids.nick_id'), primary_key=True)
    slug = Column(String(255), primary_key=True)
    canonical = Column(String(255))


class NickValues(BASE):
    """
    NickValues SQLAlchemy Class
    """
    __tablename__ = 'nick_values'
    __table_args__ = MYSQL_TABLE_ARGS
    nick_id = Column(Integer, ForeignKey('nick_ids.nick_id'), primary_key=True)
    key = Column(String(255), primary_key=True)
    value = Column(String(255))


class ChannelValues(BASE):
    """
    ChannelValues SQLAlchemy Class
    """
    __tablename__ = 'channel_values'
    __table_args__ = MYSQL_TABLE_ARGS
    channel = Column(String(255), primary_key=True)
    key = Column(String(255), primary_key=True)
    value = Column(String(255))


class PluginValues(BASE):
    """
    PluginValues SQLAlchemy Class
    """
    __tablename__ = 'plugin_values'
    __table_args__ = MYSQL_TABLE_ARGS
    plugin = Column(String(255), primary_key=True)
    key = Column(String(255), primary_key=True)
    value = Column(String(255))


class SopelDB(object):
    """*Availability: 5.0+*

    This defines an interface for basic, common operations on a sqlite
    database. It simplifies those common operations, and allows direct access
    to the database, wherever the user has configured it to be.

    When configured with a relative filename, it is assumed to be in the directory
    set (or defaulted to) in the core setting ``homedir``.
    """

    def __init__(self, config):
        # MySQL - mysql://username:password@localhost/db
        # SQLite - sqlite:////home/sopel/.sopel/default.db
        db_type = config.core.db_type

        # Handle SQLite explicitly as a default
        if db_type == 'sqlite':
            path = config.core.db_filename
            if path is None:
                path = os.path.join(config.core.homedir, config.basename + '.db')
            path = os.path.expanduser(path)
            if not os.path.isabs(path):
                path = os.path.normpath(os.path.join(config.core.homedir, path))
            if not os.path.isdir(os.path.dirname(path)):
                raise OSError(
                    errno.ENOENT,
                    'Cannot create database file. '
                    'No such directory: "{}". Check that configuration setting '
                    'core.db_filename is valid'.format(os.path.dirname(path)),
                    path
                )
            self.filename = path
            self.url = 'sqlite:///%s' % path
        # Otherwise, handle all other database engines
        else:
            query = {}
            if db_type == 'mysql':
                drivername = config.core.db_driver or 'mysql'
                query = {'charset': 'utf8mb4'}
            elif db_type == 'postgres':
                drivername = config.core.db_driver or 'postgresql'
            elif db_type == 'oracle':
                drivername = config.core.db_driver or 'oracle'
            elif db_type == 'mssql':
                drivername = config.core.db_driver or 'mssql+pymssql'
            elif db_type == 'firebird':
                drivername = config.core.db_driver or 'firebird+fdb'
            elif db_type == 'sybase':
                drivername = config.core.db_driver or 'sybase+pysybase'
            else:
                raise Exception('Unknown db_type')

            db_user = config.core.db_user
            db_pass = config.core.db_pass
            db_host = config.core.db_host
            db_port = config.core.db_port  # Optional
            db_name = config.core.db_name  # Optional, depending on DB

            # Ensure we have all our variables defined
            if db_user is None or db_pass is None or db_host is None:
                raise Exception('Please make sure the following core '
                                'configuration values are defined: '
                                'db_user, db_pass, db_host')
            self.url = URL(drivername=drivername, username=db_user,
                           password=db_pass, host=db_host, port=db_port,
                           database=db_name, query=query)

        self.engine = create_engine(self.url)

        # Catch any errors connecting to database
        try:
            self.engine.connect()
        except OperationalError:
            print("OperationalError: Unable to connect to database.")
            raise

        # Create our tables
        BASE.metadata.create_all(self.engine)

        self.ssession = scoped_session(sessionmaker(bind=self.engine))

    def connect(self):
        """Return a raw database connection object."""
        return self.engine.connect()

    def execute(self, *args, **kwargs):
        """Execute an arbitrary SQL query against the database.

        Returns a cursor object, on which things like `.fetchall()` can be
        called per PEP 249."""
        with self.connect() as conn:
            return conn.execute(*args, **kwargs)

    def get_uri(self):
        """Returns a URL for the database, usable to connect with SQLAlchemy."""
        return 'sqlite:///{}'.format(self.filename)

    # NICK FUNCTIONS

    def get_nick_id(self, nick, create=True):
        """Return the internal identifier for a given nick.

        This identifier is unique to a user, and shared across all of that
        user's aliases. If create is True, a new ID will be created if one does
        not already exist"""
        session = self.ssession()
        slug = nick.lower()
        try:
            nickname = session.query(Nicknames) \
                .filter(Nicknames.slug == slug) \
                .one_or_none()

            if nickname is None:
                if not create:
                    raise ValueError('No ID exists for the given nick')
                # Generate a new ID
                nick_id = NickIDs()
                session.add(nick_id)
                session.commit()

                # Create a new Nickname
                nickname = Nicknames(nick_id=nick_id.nick_id, slug=slug, canonical=nick)
                session.add(nickname)
                session.commit()
            return nickname.nick_id
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def alias_nick(self, nick, alias):
        """Create an alias for a nick.

        Raises ValueError if the alias already exists. If nick does not already
        exist, it will be added along with the alias."""
        nick = Identifier(nick)
        alias = Identifier(alias)
        nick_id = self.get_nick_id(nick)
        session = self.ssession()
        try:
            result = session.query(Nicknames) \
                .filter(Nicknames.slug == alias.lower()) \
                .filter(Nicknames.canonical == alias) \
                .one_or_none()
            if result:
                raise ValueError('Given alias is the only entry in its group.')
            nickname = Nicknames(nick_id=nick_id, slug=alias.lower(), canonical=alias)
            session.add(nickname)
            session.commit()
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def set_nick_value(self, nick, key, value):
        """Sets the value for a given key to be associated with the nick."""
        nick = Identifier(nick)
        value = json.dumps(value, ensure_ascii=False)
        nick_id = self.get_nick_id(nick)
        session = self.ssession()
        try:
            result = session.query(NickValues) \
                .filter(NickValues.nick_id == nick_id) \
                .filter(NickValues.key == key) \
                .one_or_none()
            # NickValue exists, update
            if result:
                result.value = value
                session.commit()
            # DNE - Insert
            else:
                new_nickvalue = NickValues(nick_id=nick_id, key=key, value=value)
                session.add(new_nickvalue)
                session.commit()
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def delete_nick_value(self, nick, key):
        """Deletes the value for a given key associated with a nick."""
        nick = Identifier(nick)
        nick_id = self.get_nick_id(nick)
        session = self.ssession()
        try:
            result = session.query(NickValues) \
                .filter(NickValues.nick_id == nick_id) \
                .filter(NickValues.key == key) \
                .one_or_none()
            # NickValue exists, delete
            if result:
                session.delete(result)
                session.commit()
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def get_nick_value(self, nick, key):
        """Retrieves the value for a given key associated with a nick."""
        nick = Identifier(nick)
        session = self.ssession()
        try:
            result = session.query(NickValues) \
                .filter(Nicknames.nick_id == NickValues.nick_id) \
                .filter(Nicknames.slug == nick.lower()) \
                .filter(NickValues.key == key) \
                .one_or_none()
            if result is not None:
                result = result.value
            return _deserialize(result)
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def unalias_nick(self, alias):
        """Removes an alias.

        Raises ValueError if there is not at least one other nick in the group.
        To delete an entire group, use `delete_group`.
        """
        alias = Identifier(alias)
        nick_id = self.get_nick_id(alias, False)
        session = self.ssession()
        try:
            count = session.query(Nicknames) \
                .filter(Nicknames.nick_id == nick_id) \
                .count()
            if count <= 1:
                raise ValueError('Given alias is the only entry in its group.')
            session.query(Nicknames).filter(Nicknames.slug == alias.lower()).delete()
            session.commit()
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def delete_nick_group(self, nick):
        """Removes a nickname, and all associated aliases and settings."""
        nick = Identifier(nick)
        nick_id = self.get_nick_id(nick, False)
        session = self.ssession()
        try:
            session.query(Nicknames).filter(Nicknames.nick_id == nick_id).delete()
            session.query(NickValues).filter(NickValues.nick_id == nick_id).delete()
            session.commit()
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def merge_nick_groups(self, first_nick, second_nick):
        """Merges the nick groups for the specified nicks.

        Takes two nicks, which may or may not be registered.  Unregistered
        nicks will be registered. Keys which are set for only one of the given
        nicks will be preserved. Where multiple nicks have values for a given
        key, the value set for the first nick will be used.

        Note that merging of data only applies to the native key-value store.
        If modules define their own tables which rely on the nick table, they
        will need to have their merging done separately."""
        first_id = self.get_nick_id(Identifier(first_nick))
        second_id = self.get_nick_id(Identifier(second_nick))
        session = self.ssession()
        try:
            # Get second_id's values
            res = session.query(NickValues).filter(NickValues.nick_id == second_id).all()
            # Update first_id with second_id values if first_id doesn't have that key
            for row in res:
                first_res = session.query(NickValues) \
                    .filter(NickValues.nick_id == first_id) \
                    .filter(NickValues.key == row.key) \
                    .one_or_none()
                if not first_res:
                    self.set_nick_value(first_nick, row.key, _deserialize(row.value))
            session.query(NickValues).filter(NickValues.nick_id == second_id).delete()
            session.query(Nicknames) \
                .filter(Nicknames.nick_id == second_id) \
                .update({'nick_id': first_id})
            session.commit()
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    # CHANNEL FUNCTIONS

    def set_channel_value(self, channel, key, value):
        """Sets the value for a given key to be associated with the channel."""
        channel = Identifier(channel).lower()
        value = json.dumps(value, ensure_ascii=False)
        session = self.ssession()
        try:
            result = session.query(ChannelValues) \
                .filter(ChannelValues.channel == channel)\
                .filter(ChannelValues.key == key) \
                .one_or_none()
            # ChannelValue exists, update
            if result:
                result.value = value
                session.commit()
            # DNE - Insert
            else:
                new_channelvalue = ChannelValues(channel=channel, key=key, value=value)
                session.add(new_channelvalue)
                session.commit()
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def delete_channel_value(self, channel, key):
        """Deletes the value for a given key associated with a channel."""
        channel = Identifier(channel).lower()
        session = self.ssession()
        try:
            result = session.query(ChannelValues) \
                .filter(ChannelValues.channel == channel)\
                .filter(ChannelValues.key == key) \
                .one_or_none()
            # ChannelValue exists, delete
            if result:
                session.delete(result)
                session.commit()
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def get_channel_value(self, channel, key):
        """Retrieves the value for a given key associated with a channel."""
        channel = Identifier(channel).lower()
        session = self.ssession()
        try:
            result = session.query(ChannelValues) \
                .filter(ChannelValues.channel == channel)\
                .filter(ChannelValues.key == key) \
                .one_or_none()
            if result is not None:
                result = result.value
            return _deserialize(result)
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    # PLUGIN FUNCTIONS

    def set_plugin_value(self, plugin, key, value):
        """Sets the value for a given key to be associated with a plugin."""
        plugin = plugin.lower()
        value = json.dumps(value, ensure_ascii=False)
        session = self.ssession()
        try:
            result = session.query(PluginValues) \
                .filter(PluginValues.plugin == plugin)\
                .filter(PluginValues.key == key) \
                .one_or_none()
            # PluginValue exists, update
            if result:
                result.value = value
                session.commit()
            # DNE - Insert
            else:
                new_pluginvalue = PluginValues(plugin=plugin, key=key, value=value)
                session.add(new_pluginvalue)
                session.commit()
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def delete_plugin_value(self, plugin, key):
        """Deletes the value for a given key associated with a plugin."""
        plugin = plugin.lower()
        session = self.ssession()
        try:
            result = session.query(PluginValues) \
                .filter(PluginValues.plugin == plugin)\
                .filter(PluginValues.key == key) \
                .one_or_none()
            # PluginValue exists, update
            if result:
                session.delete(result)
                session.commit()
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    def get_plugin_value(self, plugin, key):
        """Retrieves the value for a given key associated with a plugin."""
        plugin = plugin.lower()
        session = self.ssession()
        try:
            result = session.query(PluginValues) \
                .filter(PluginValues.plugin == plugin)\
                .filter(PluginValues.key == key) \
                .one_or_none()
            if result is not None:
                result = result.value
            return _deserialize(result)
        except SQLAlchemyError:
            session.rollback()
            raise
        finally:
            session.close()

    # NICK AND CHANNEL FUNCTIONS

    def get_nick_or_channel_value(self, name, key):
        """Gets the value `key` associated to the nick or channel  `name`."""
        name = Identifier(name)
        if name.is_nick():
            return self.get_nick_value(name, key)
        else:
            return self.get_channel_value(name, key)

    def get_preferred_value(self, names, key):
        """Gets the value for the first name which has it set.

        `names` is a list of channel and/or user names. Returns None if none of
        the names have the key set."""
        for name in names:
            value = self.get_nick_or_channel_value(name, key)
            if value is not None:
                return value


1			# coding=utf-8
2			from __future__ import unicode_literals, absolute_import, print_function, division
3
4			import errno
5			import json
6			import os.path
7			import sys
8
9			from sopel.tools import Identifier
10
11			from sqlalchemy import create_engine, Column, ForeignKey, Integer, String
12			from sqlalchemy.engine.url import URL
13			from sqlalchemy.exc import OperationalError, SQLAlchemyError
14			from sqlalchemy.ext.declarative import declarative_base
15			from sqlalchemy.orm import scoped_session, sessionmaker
16
17			if sys.version_info.major >= 3:
18			unicode = str
19			basestring = str
20
21
22			def _deserialize(value):
23			if value is None:
24			return None
25			# sqlite likes to return ints for strings that look like ints, even though
26			# the column type is string. That's how you do dynamic typing wrong.
27			value = unicode(value)
28			# Just in case someone's mucking with the DB in a way we can't account for,
29			# ignore json parsing errors
30			try:
31			value = json.loads(value)
32			except ValueError:
33			pass
34			return value
35
36
37			BASE = declarative_base()
38			MYSQL_TABLE_ARGS = {'mysql_engine': 'InnoDB',
39			'mysql_charset': 'utf8mb4',
40			'mysql_collate': 'utf8mb4_unicode_ci'}
41
42
43			class NickIDs(BASE):
44			"""
45			NickIDs SQLAlchemy Class
46			"""
47			__tablename__ = 'nick_ids'
48			nick_id = Column(Integer, primary_key=True)
49
50
51			class Nicknames(BASE):
52			"""
53			Nicknames SQLAlchemy Class
54			"""
55			__tablename__ = 'nicknames'
56			__table_args__ = MYSQL_TABLE_ARGS
57			nick_id = Column(Integer, ForeignKey('nick_ids.nick_id'), primary_key=True)
58			slug = Column(String(255), primary_key=True)
59			canonical = Column(String(255))
60
61
62			class NickValues(BASE):
63			"""
64			NickValues SQLAlchemy Class
65			"""
66			__tablename__ = 'nick_values'
67			__table_args__ = MYSQL_TABLE_ARGS
68			nick_id = Column(Integer, ForeignKey('nick_ids.nick_id'), primary_key=True)
69			key = Column(String(255), primary_key=True)
70			value = Column(String(255))
71
72
73			class ChannelValues(BASE):
74			"""
75			ChannelValues SQLAlchemy Class
76			"""
77			__tablename__ = 'channel_values'
78			__table_args__ = MYSQL_TABLE_ARGS
79			channel = Column(String(255), primary_key=True)
80			key = Column(String(255), primary_key=True)
81			value = Column(String(255))
82
83
84			class PluginValues(BASE):
85			"""
86			PluginValues SQLAlchemy Class
87			"""
88			__tablename__ = 'plugin_values'
89			__table_args__ = MYSQL_TABLE_ARGS
90			plugin = Column(String(255), primary_key=True)
91			key = Column(String(255), primary_key=True)
92			value = Column(String(255))
93
94
95			class SopelDB(object):
96			"""Availability: 5.0+
97
98			This defines an interface for basic, common operations on a sqlite
99			database. It simplifies those common operations, and allows direct access
100			to the database, wherever the user has configured it to be.
101
102			When configured with a relative filename, it is assumed to be in the directory
103			set (or defaulted to) in the core setting ``homedir``.
104			"""
105
106			def __init__(self, config):
107			# MySQL - mysql://username:password@localhost/db
108			# SQLite - sqlite:////home/sopel/.sopel/default.db
109			db_type = config.core.db_type
110
111			# Handle SQLite explicitly as a default
112			if db_type == 'sqlite':
113			path = config.core.db_filename
114			if path is None:
115			path = os.path.join(config.core.homedir, config.basename + '.db')
116			path = os.path.expanduser(path)
117			if not os.path.isabs(path):
118			path = os.path.normpath(os.path.join(config.core.homedir, path))
119			if not os.path.isdir(os.path.dirname(path)):
120			raise OSError(
121			errno.ENOENT,
122			'Cannot create database file. '
123			'No such directory: "{}". Check that configuration setting '
124			'core.db_filename is valid'.format(os.path.dirname(path)),
125			path
126			)
127			self.filename = path
128			self.url = 'sqlite:///%s' % path
129			# Otherwise, handle all other database engines
130			else:
131			query = {}
132			if db_type == 'mysql':
133			drivername = config.core.db_driver or 'mysql'
134			query = {'charset': 'utf8mb4'}
135			elif db_type == 'postgres':
136			drivername = config.core.db_driver or 'postgresql'
137			elif db_type == 'oracle':
138			drivername = config.core.db_driver or 'oracle'
139			elif db_type == 'mssql':
140			drivername = config.core.db_driver or 'mssql+pymssql'
141			elif db_type == 'firebird':
142			drivername = config.core.db_driver or 'firebird+fdb'
143			elif db_type == 'sybase':
144			drivername = config.core.db_driver or 'sybase+pysybase'
145			else:
146			raise Exception('Unknown db_type')
147
148			db_user = config.core.db_user
149			db_pass = config.core.db_pass
150			db_host = config.core.db_host
151			db_port = config.core.db_port # Optional
152			db_name = config.core.db_name # Optional, depending on DB
153
154			# Ensure we have all our variables defined
155			if db_user is None or db_pass is None or db_host is None:
156			raise Exception('Please make sure the following core '
157			'configuration values are defined: '
158			'db_user, db_pass, db_host')
159			self.url = URL(drivername=drivername, username=db_user,
160			password=db_pass, host=db_host, port=db_port,
161			database=db_name, query=query)
162
163			self.engine = create_engine(self.url)
164
165			# Catch any errors connecting to database
166			try:
167			self.engine.connect()
168			except OperationalError:
169			print("OperationalError: Unable to connect to database.")
170			raise
171
172			# Create our tables
173			BASE.metadata.create_all(self.engine)
174
175			self.ssession = scoped_session(sessionmaker(bind=self.engine))
176
177			def connect(self):
178			"""Return a raw database connection object."""
179			return self.engine.connect()
180
181			def execute(self, args, *kwargs):
182			"""Execute an arbitrary SQL query against the database.
183
184			Returns a cursor object, on which things like `.fetchall()` can be
185			called per PEP 249."""
186			with self.connect() as conn:
187			return conn.execute(args, *kwargs)
188
189			def get_uri(self):
190			"""Returns a URL for the database, usable to connect with SQLAlchemy."""
191			return 'sqlite:///{}'.format(self.filename)
192
193			# NICK FUNCTIONS
194
195			def get_nick_id(self, nick, create=True):
196			"""Return the internal identifier for a given nick.
197
198			This identifier is unique to a user, and shared across all of that
199			user's aliases. If create is True, a new ID will be created if one does
200			not already exist"""
201			session = self.ssession()
202			slug = nick.lower()
203			try:
204			nickname = session.query(Nicknames) \
205			.filter(Nicknames.slug == slug) \
206			.one_or_none()
207
208			if nickname is None:
209			if not create:
210			raise ValueError('No ID exists for the given nick')
211			# Generate a new ID
212			nick_id = NickIDs()
213			session.add(nick_id)
214			session.commit()
215
216			# Create a new Nickname
217			nickname = Nicknames(nick_id=nick_id.nick_id, slug=slug, canonical=nick)
218			session.add(nickname)
219			session.commit()
220			return nickname.nick_id
221			except SQLAlchemyError:
222			session.rollback()
223			raise
224			finally:
225			session.close()
226
227			def alias_nick(self, nick, alias):
228			"""Create an alias for a nick.
229
230			Raises ValueError if the alias already exists. If nick does not already
231			exist, it will be added along with the alias."""
232			nick = Identifier(nick)
233			alias = Identifier(alias)
234			nick_id = self.get_nick_id(nick)
235			session = self.ssession()
236			try:
237			result = session.query(Nicknames) \
238			.filter(Nicknames.slug == alias.lower()) \
239			.filter(Nicknames.canonical == alias) \
240			.one_or_none()
241			if result:
242			raise ValueError('Given alias is the only entry in its group.')
243			nickname = Nicknames(nick_id=nick_id, slug=alias.lower(), canonical=alias)
244			session.add(nickname)
245			session.commit()
246			except SQLAlchemyError:
247			session.rollback()
248			raise
249			finally:
250			session.close()
251
252			def set_nick_value(self, nick, key, value):
253			"""Sets the value for a given key to be associated with the nick."""
254			nick = Identifier(nick)
255			value = json.dumps(value, ensure_ascii=False)
256			nick_id = self.get_nick_id(nick)
257			session = self.ssession()
258			try:
259			result = session.query(NickValues) \
260			.filter(NickValues.nick_id == nick_id) \
261			.filter(NickValues.key == key) \
262			.one_or_none()
263			# NickValue exists, update
264			if result:
265			result.value = value
266			session.commit()
267			# DNE - Insert
268			else:
269			new_nickvalue = NickValues(nick_id=nick_id, key=key, value=value)
270			session.add(new_nickvalue)
271			session.commit()
272			except SQLAlchemyError:
273			session.rollback()
274			raise
275			finally:
276			session.close()
277
278			def delete_nick_value(self, nick, key):
279			"""Deletes the value for a given key associated with a nick."""
280			nick = Identifier(nick)
281			nick_id = self.get_nick_id(nick)
282			session = self.ssession()
283			try:
284			result = session.query(NickValues) \
285			.filter(NickValues.nick_id == nick_id) \
286			.filter(NickValues.key == key) \
287			.one_or_none()
288			# NickValue exists, delete
289			if result:
290			session.delete(result)
291			session.commit()
292			except SQLAlchemyError:
293			session.rollback()
294			raise
295			finally:
296			session.close()
297
298			def get_nick_value(self, nick, key):
299			"""Retrieves the value for a given key associated with a nick."""
300			nick = Identifier(nick)
301			session = self.ssession()
302			try:
303			result = session.query(NickValues) \
304			.filter(Nicknames.nick_id == NickValues.nick_id) \
305			.filter(Nicknames.slug == nick.lower()) \
306			.filter(NickValues.key == key) \
307			.one_or_none()
308			if result is not None:
309			result = result.value
310			return _deserialize(result)
311			except SQLAlchemyError:
312			session.rollback()
313			raise
314			finally:
315			session.close()
316
317			def unalias_nick(self, alias):
318			"""Removes an alias.
319
320			Raises ValueError if there is not at least one other nick in the group.
321			To delete an entire group, use `delete_group`.
322			"""
323			alias = Identifier(alias)
324			nick_id = self.get_nick_id(alias, False)
325			session = self.ssession()
326			try:
327			count = session.query(Nicknames) \
328			.filter(Nicknames.nick_id == nick_id) \
329			.count()
330			if count <= 1:
331			raise ValueError('Given alias is the only entry in its group.')
332			session.query(Nicknames).filter(Nicknames.slug == alias.lower()).delete()
333			session.commit()
334			except SQLAlchemyError:
335			session.rollback()
336			raise
337			finally:
338			session.close()
339
340			def delete_nick_group(self, nick):
341			"""Removes a nickname, and all associated aliases and settings."""
342			nick = Identifier(nick)
343			nick_id = self.get_nick_id(nick, False)
344			session = self.ssession()
345			try:
346			session.query(Nicknames).filter(Nicknames.nick_id == nick_id).delete()
347			session.query(NickValues).filter(NickValues.nick_id == nick_id).delete()
348			session.commit()
349			except SQLAlchemyError:
350			session.rollback()
351			raise
352			finally:
353			session.close()
354
355			def merge_nick_groups(self, first_nick, second_nick):
356			"""Merges the nick groups for the specified nicks.
357
358			Takes two nicks, which may or may not be registered. Unregistered
359			nicks will be registered. Keys which are set for only one of the given
360			nicks will be preserved. Where multiple nicks have values for a given
361			key, the value set for the first nick will be used.
362
363			Note that merging of data only applies to the native key-value store.
364			If modules define their own tables which rely on the nick table, they
365			will need to have their merging done separately."""
366			first_id = self.get_nick_id(Identifier(first_nick))
367			second_id = self.get_nick_id(Identifier(second_nick))
368			session = self.ssession()
369			try:
370			# Get second_id's values
371			res = session.query(NickValues).filter(NickValues.nick_id == second_id).all()
372			# Update first_id with second_id values if first_id doesn't have that key
373			for row in res:
374			first_res = session.query(NickValues) \
375			.filter(NickValues.nick_id == first_id) \
376			.filter(NickValues.key == row.key) \
377			.one_or_none()
378			if not first_res:
379			self.set_nick_value(first_nick, row.key, _deserialize(row.value))
380			session.query(NickValues).filter(NickValues.nick_id == second_id).delete()
381			session.query(Nicknames) \
382			.filter(Nicknames.nick_id == second_id) \
383			.update({'nick_id': first_id})
384			session.commit()
385			except SQLAlchemyError:
386			session.rollback()
387			raise
388			finally:
389			session.close()
390
391			# CHANNEL FUNCTIONS
392
393			def set_channel_value(self, channel, key, value):
394			"""Sets the value for a given key to be associated with the channel."""
395			channel = Identifier(channel).lower()
396			value = json.dumps(value, ensure_ascii=False)
397			session = self.ssession()
398			try:
399			result = session.query(ChannelValues) \
400			.filter(ChannelValues.channel == channel)\
401			.filter(ChannelValues.key == key) \
402			.one_or_none()
403			# ChannelValue exists, update
404			if result:
405			result.value = value
406			session.commit()
407			# DNE - Insert
408			else:
409			new_channelvalue = ChannelValues(channel=channel, key=key, value=value)
410			session.add(new_channelvalue)
411			session.commit()
412			except SQLAlchemyError:
413			session.rollback()
414			raise
415			finally:
416			session.close()
417
418			def delete_channel_value(self, channel, key):
419			"""Deletes the value for a given key associated with a channel."""
420			channel = Identifier(channel).lower()
421			session = self.ssession()
422			try:
423			result = session.query(ChannelValues) \
424			.filter(ChannelValues.channel == channel)\
425			.filter(ChannelValues.key == key) \
426			.one_or_none()
427			# ChannelValue exists, delete
428			if result:
429			session.delete(result)
430			session.commit()
431			except SQLAlchemyError:
432			session.rollback()
433			raise
434			finally:
435			session.close()
436
437			def get_channel_value(self, channel, key):
438			"""Retrieves the value for a given key associated with a channel."""
439			channel = Identifier(channel).lower()
440			session = self.ssession()
441			try:
442			result = session.query(ChannelValues) \
443			.filter(ChannelValues.channel == channel)\
444			.filter(ChannelValues.key == key) \
445			.one_or_none()
446			if result is not None:
447			result = result.value
448			return _deserialize(result)
449			except SQLAlchemyError:
450			session.rollback()
451			raise
452			finally:
453			session.close()
454
455			# PLUGIN FUNCTIONS
456
457			def set_plugin_value(self, plugin, key, value):
458			"""Sets the value for a given key to be associated with a plugin."""
459			plugin = plugin.lower()
460			value = json.dumps(value, ensure_ascii=False)
461			session = self.ssession()
462			try:
463			result = session.query(PluginValues) \
464			.filter(PluginValues.plugin == plugin)\
465			.filter(PluginValues.key == key) \
466			.one_or_none()
467			# PluginValue exists, update
468			if result:
469			result.value = value
470			session.commit()
471			# DNE - Insert
472			else:
473			new_pluginvalue = PluginValues(plugin=plugin, key=key, value=value)
474			session.add(new_pluginvalue)
475			session.commit()
476			except SQLAlchemyError:
477			session.rollback()
478			raise
479			finally:
480			session.close()
481
482			def delete_plugin_value(self, plugin, key):
483			"""Deletes the value for a given key associated with a plugin."""
484			plugin = plugin.lower()
485			session = self.ssession()
486			try:
487			result = session.query(PluginValues) \
488			.filter(PluginValues.plugin == plugin)\
489			.filter(PluginValues.key == key) \
490			.one_or_none()
491			# PluginValue exists, update
492			if result:
493			session.delete(result)
494			session.commit()
495			except SQLAlchemyError:
496			session.rollback()
497			raise
498			finally:
499			session.close()
500
501			def get_plugin_value(self, plugin, key):
502			"""Retrieves the value for a given key associated with a plugin."""
503			plugin = plugin.lower()
504			session = self.ssession()
505			try:
506			result = session.query(PluginValues) \
507			.filter(PluginValues.plugin == plugin)\
508			.filter(PluginValues.key == key) \
509			.one_or_none()
510			if result is not None:
511			result = result.value
512			return _deserialize(result)
513			except SQLAlchemyError:
514			session.rollback()
515			raise
516			finally:
517			session.close()
518
519			# NICK AND CHANNEL FUNCTIONS
520
521			def get_nick_or_channel_value(self, name, key):
522			"""Gets the value `key` associated to the nick or channel `name`."""
523			name = Identifier(name)
524			if name.is_nick():
525			return self.get_nick_value(name, key)
526			else:
527			return self.get_channel_value(name, key)
528
529			def get_preferred_value(self, names, key):
530			"""Gets the value for the first name which has it set.
531
532			`names` is a list of channel and/or user names. Returns None if none of
533			the names have the key set."""
534			for name in names:
535			value = self.get_nick_or_channel_value(name, key)
536			if value is not None:
537			return value
538

sopel-irc / sopel

sopel.db.SopelDB.__init__() F last analyzed 2019-10-21 18:25 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

sopel.db.SopelDB.init() F
last analyzed 2019-10-21 18:25 UTC