from packaging import version
import warnings

from . import CollectionInvalid
from . import InvalidName
from . import OperationFailure
from .collection import Collection
from .filtering import filter_applies
from mongomock import codec_options as mongomock_codec_options
from mongomock import helpers
from mongomock import read_preferences
from mongomock import store

try:
    from pymongo import ReadPreference
    _READ_PREFERENCE_PRIMARY = ReadPreference.PRIMARY
except ImportError:
    _READ_PREFERENCE_PRIMARY = read_preferences.PRIMARY

try:
    from pymongo.read_concern import ReadConcern
except ImportError:
    from .read_concern import ReadConcern

_LIST_COLLECTION_FILTER_ALLOWED_OPERATORS = frozenset(['$regex', '$eq', '$ne'])


def _verify_list_collection_supported_op(keys):
    if set(keys) - _LIST_COLLECTION_FILTER_ALLOWED_OPERATORS:
        raise NotImplementedError(
            'list collection names filter operator {} is not implemented yet in mongomock '
            'allowed operators are {}'.format(keys, _LIST_COLLECTION_FILTER_ALLOWED_OPERATORS))


class Database:

    def __init__(
        self, client, name, _store, read_preference=None, codec_options=None, read_concern=None
    ):
        self.name = name
        self._client = client
        self._collection_accesses = {}
        self._store = _store or store.DatabaseStore()
        self._read_preference = read_preference or _READ_PREFERENCE_PRIMARY
        mongomock_codec_options.is_supported(codec_options)
        self._codec_options = codec_options or mongomock_codec_options.CodecOptions()
        if read_concern and not isinstance(read_concern, ReadConcern):
            raise TypeError('read_concern must be an instance of pymongo.read_concern.ReadConcern')
        self._read_concern = read_concern or ReadConcern()

    def __getitem__(self, coll_name):
        return self.get_collection(coll_name)

    def __getattr__(self, attr):
        if attr.startswith('_'):
            raise AttributeError(
                "%s has no attribute '%s'. To access the %s collection, use database['%s']." %
                (self.__class__.__name__, attr, attr, attr))
        return self[attr]

    def __repr__(self):
        return f"Database({self._client}, '{self.name}')"

    def __eq__(self, other):
        if isinstance(other, self.__class__):
            return self._client == other._client and self.name == other.name
        return NotImplemented

    if helpers.PYMONGO_VERSION >= version.parse('3.12'):
        def __hash__(self):
            return hash((self._client, self.name))

    @property
    def client(self):
        return self._client

    @property
    def read_preference(self):
        return self._read_preference

    @property
    def codec_options(self):
        return self._codec_options

    @property
    def read_concern(self):
        return self._read_concern

    def _get_created_collections(self):
        return self._store.list_created_collection_names()

    if helpers.PYMONGO_VERSION < version.parse('4.0'):
        def collection_names(self, include_system_collections=True, session=None):
            warnings.warn('collection_names is deprecated. Use list_collection_names instead.')
            if include_system_collections:
                return list(self._get_created_collections())
            return self.list_collection_names(session=session)

    def list_collections(self, filter=None, session=None, nameOnly=False):
        raise NotImplementedError(
            'list_collections is a valid method of Database but has not been implemented in '
            'mongomock yet.')

    def list_collection_names(self, filter=None, session=None):
        """filter: only name field type with eq,ne or regex operator

        session: not supported
        for supported operator please see _LIST_COLLECTION_FILTER_ALLOWED_OPERATORS
        """
        field_name = 'name'

        if session:
            raise NotImplementedError('Mongomock does not handle sessions yet')

        if filter:
            if not filter.get('name'):
                raise NotImplementedError('list collection {} might be valid but is not '
                                          'implemented yet in mongomock'.format(filter))

            filter = {field_name: {'$eq': filter.get(field_name)}} \
                if isinstance(filter.get(field_name), str) else filter

            _verify_list_collection_supported_op(filter.get(field_name).keys())

            return [
                name for name in list(self._store._collections)
                if filter_applies(filter, {field_name: name}) and not name.startswith('system.')
            ]

        return [
            name for name in self._get_created_collections()
            if not name.startswith('system.')
        ]

    def get_collection(self, name, codec_options=None, read_preference=None,
                       write_concern=None, read_concern=None):
        if read_preference is not None:
            read_preferences.ensure_read_preference_type('read_preference', read_preference)
        mongomock_codec_options.is_supported(codec_options)
        try:
            return self._collection_accesses[name].with_options(
                codec_options=codec_options or self._codec_options,
                read_preference=read_preference or self.read_preference,
                read_concern=read_concern, write_concern=write_concern)
        except KeyError:
            self._ensure_valid_collection_name(name)
            collection = self._collection_accesses[name] = Collection(
                self, name=name, read_concern=read_concern, write_concern=write_concern,
                read_preference=read_preference or self.read_preference,
                codec_options=codec_options or self._codec_options, _db_store=self._store, )
            return collection

    def drop_collection(self, name_or_collection, session=None):
        if session:
            raise NotImplementedError('Mongomock does not handle sessions yet')
        if isinstance(name_or_collection, Collection):
            name_or_collection._store.drop()
        else:
            self._store[name_or_collection].drop()

    def _ensure_valid_collection_name(self, name):
        # These are the same checks that are done in pymongo.
        if not isinstance(name, str):
            raise TypeError('name must be an instance of str')
        if not name or '..' in name:
            raise InvalidName('collection names cannot be empty')
        if name[0] == '.' or name[-1] == '.':
            raise InvalidName("collection names must not start or end with '.'")
        if '$' in name:
            raise InvalidName("collection names must not contain '$'")
        if '\x00' in name:
            raise InvalidName('collection names must not contain the null character')

    def create_collection(self, name, **kwargs):
        self._ensure_valid_collection_name(name)
        if name in self.list_collection_names():
            raise CollectionInvalid('collection %s already exists' % name)

        if kwargs:
            raise NotImplementedError('Special options not supported')

        self._store.create_collection(name)
        return self[name]

    def rename_collection(self, name, new_name, dropTarget=False):
        """Changes the name of an existing collection."""
        self._ensure_valid_collection_name(new_name)

        # Reference for server implementation:
        # https://docs.mongodb.com/manual/reference/command/renameCollection/
        if not self._store[name].is_created:
            raise OperationFailure(
                f'The collection "{name}" does not exist.', 10026)
        if new_name in self._store:
            if dropTarget:
                self.drop_collection(new_name)
            else:
                raise OperationFailure(
                    f'The target collection "{new_name}" already exists',
                    10027)
        self._store.rename(name, new_name)
        return {'ok': 1}

    def dereference(self, dbref, session=None):
        if session:
            raise NotImplementedError('Mongomock does not handle sessions yet')

        if not hasattr(dbref, 'collection') or not hasattr(dbref, 'id'):
            raise TypeError('cannot dereference a %s' % type(dbref))
        if dbref.database is not None and dbref.database != self.name:
            raise ValueError('trying to dereference a DBRef that points to '
                             'another database (%r not %r)' % (dbref.database,
                                                               self.name))
        return self[dbref.collection].find_one({'_id': dbref.id})

    def command(self, command, **unused_kwargs):
        if isinstance(command, str):
            command = {command: 1}
        if 'ping' in command:
            return {'ok': 1.}
        # TODO(pascal): Differentiate NotImplementedError for valid commands
        # and OperationFailure if the command is not valid.
        raise NotImplementedError(
            'command is a valid Database method but is not implemented in Mongomock yet')

    def with_options(
            self, codec_options=None, read_preference=None, write_concern=None, read_concern=None):

        mongomock_codec_options.is_supported(codec_options)

        if write_concern:
            raise NotImplementedError(
                'write_concern is a valid parameter for with_options but is not implemented yet in '
                'mongomock')

        if read_preference is None or read_preference == self._read_preference:
            return self

        return Database(
            self._client, self.name, self._store,
            read_preference=read_preference or self._read_preference,
            codec_options=codec_options or self._codec_options,
            read_concern=read_concern or self._read_concern,
        )