from . import core

from .. import config as configmod

from .. import data

from .. import errors as errormod

from ..beancount_types import (

    MetaValueEnum,

    Transaction,

)

from typing import (

    Any,

    Dict,

    Optional,

    Set,

)

class MetaProject(core._NormalizePostingMetadataHook):

    DEFAULT_PROJECT = 'Conservancy'

    PROJECT_DATA_PATH = Path('Projects', 'project-data.yml')

    VALUES_ENUM = core.MetadataEnum('project', {DEFAULT_PROJECT})

    def __init__(self, config: configmod.Config, source_path: Path=PROJECT_DATA_PATH) -> None:

        repo_path = config.repository_path()

        if repo_path is None:

        project_data_path = repo_path / source_path

        source = {'filename': str(project_data_path)}

        try:

            with project_data_path.open() as yaml_file:

                project_data: Dict[str, Dict[str, Any]] = yaml.safe_load(yaml_file)

            names: Set[MetaValueEnum] = {self.DEFAULT_PROJECT}

            aliases: Dict[MetaValueEnum, MetaValueEnum] = {}

            for key, params in project_data.items():

                name = params.get('accountName', key)

                names.add(name)

                human_name = params.get('humanName', name)

                if name != human_name:

                    aliases[human_name] = name

                if name != key:

                    aliases[key] = name

        except AttributeError:

            self._config_error("loaded YAML data not in project-data format", project_data_path)

        except OSError as error:

            self._config_error(error.strerror, project_data_path)

        except yaml.error.YAMLError as error:

            self._config_error(error.args[0] or "YAML load error", project_data_path)

        else:

            self.VALUES_ENUM = core.MetadataEnum(self.METADATA_KEY, names, aliases)

import functools

import mimetypes

import re

import sqlite3

import urllib.parse as urlparse

import rt

from pathlib import Path

from typing import (

    Callable,

    Iterator,

    MutableMapping,

    Optional,

    Set,

    Tuple,

    Union,

)

RTId = Union[int, str]

TicketAttachmentIds = Tuple[str, Optional[str]]

_LinkCache = MutableMapping[TicketAttachmentIds, Optional[str]]

class RTLinkCache(_LinkCache):

    """Cache RT links to disk

    This class provides a dict-like interface to a cache of RT links.

    Once an object is in RT, a link to it should never change.

    The only exception is when objects get shredded, and those objects

    shouldn't be referenced in books anyway.

    This implementation is backed by a sqlite database. You can call::

        db = RTLinkCache.setup(path)

    This method will try to open a sqlite database at the given path,

    and set up necessary tables, etc.

    If it succeeds, it returns a database connection you can use to

    initialize the cache.

    If it fails, it returns None, and the caller should use some other

    dict-like object (like a normal dict) for caching.

    You can give the result to the RT utility class either way,

    and it will do the right thing for itself::

        rt = RT(rt_client, db)

    """

                cursor = db.cursor()

                cursor.execute('ATTACH DATABASE ? AS readsource',

                               ('{}?mode=ro'.format(cache_path.as_uri()),))

                cursor.execute(cls.CREATE_TABLE_SQL)

                cursor.execute('INSERT INTO RTLinkCache SELECT * FROM readsource.RTLinkCache')

                cursor.execute('DETACH DATABASE readsource')

            except sqlite3.OperationalError as error:

                # We're back to the case of having nothing to read and no way

                # to write.

                return None

        cursor.close()

        db.commit()

        return db

    def __init__(self, cache_db: sqlite3.Connection) -> None:

        self._db = cache_db

        self._nourls: Set[TicketAttachmentIds] = set()

    def __iter__(self) -> Iterator[TicketAttachmentIds]:

        yield from self._db.execute('SELECT ticket_id, attachment_id FROM RTLinkCache')

        yield from self._nourls

    def __len__(self) -> int:

        cursor = self._db.execute('SELECT COUNT(*) FROM RTLinkCache')

    def __getitem__(self, key: TicketAttachmentIds) -> Optional[str]:

        if key in self._nourls:

            return None

        cursor = self._db.execute(

            'SELECT url FROM RTLinkCache WHERE ticket_id = ? AND attachment_id IS ?',

            key,

        )

            raise KeyError(key)

        else:

    def __setitem__(self, key: TicketAttachmentIds, value: Optional[str]) -> None:

        if value is None:

            self._nourls.add(key)

        else:

            ticket_id, attachment_id = key

            self._db.execute(

                'INSERT INTO RTLinkCache VALUES(?, ?, ?)',

                (ticket_id, attachment_id, value),

            )

    def __delitem__(self, key: TicketAttachmentIds) -> None:

        raise NotImplementedError("RTLinkCache.__delitem__")

class RT:

    """RT utility wrapper class

    Given an RT client object, this class provides common functionality for

    working with RT links in Beancount metadata:

    * Parse links

    * Verify that they refer to extant objects in RT

    * Convert metadata links to RT web links

    PARSE_REGEXPS = [

        re.compile(r'^rt:([0-9]+)(?:/([0-9]+))?/?$'),

        re.compile(r'^rt://ticket/([0-9]+)(?:/attachments?/([0-9]+))?/?$'),

    ]

    def __init__(self, rt_client: rt.Rt, cache_db: Optional[sqlite3.Connection]=None) -> None:

        urlparts = urlparse.urlparse(rt_client.url)

        try:

            index = urlparts.path.rindex('/REST/')

        except ValueError:

            base_path = urlparts.path.rstrip('/') + '/'

        else:

            base_path = urlparts.path[:index + 1]

        self.url_base = urlparts._replace(path=base_path)

        self.rt = rt_client

        self._cache: _LinkCache

        if cache_db is None:

            self._cache = {}

        else:

            self._cache = RTLinkCache(cache_db)

    # mypy complains that the first argument isn't self, but this isn't meant

    # to be a method, it's just an internal decrator.

        @functools.wraps(func)

        def caching_wrapper(self,

                            ticket_id: RTId,

                            attachment_id: Optional[RTId]=None,

            cache_key = (str(ticket_id), attachment_id and str(attachment_id))

            try:

                url = self._cache[cache_key]

            except KeyError:

                if attachment_id is None:

                    url = func(self, ticket_id)

                else:

                    url = func(self, ticket_id, attachment_id)

                self._cache[cache_key] = url

            return url

        return caching_wrapper

    def _extend_url(self,

                    path_tail: str,

                    fragment: Optional[str]=None,

                    **query: str,

    ) -> str:

        if fragment is None:

            fragment = self.url_base.fragment

        else:

            fragment = urlparse.quote(fragment)

        if query:

            query_s = urlparse.urlencode(query)

        else:

            query_s = self.url_base.query

[aliases]

test=pytest

typecheck=pytest --addopts="--mypy conservancy_beancount"

[mypy]

show_error_codes = True

warn_unused_configs = True