KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use repo_id method in Kiwi

Best Python code snippet using Kiwi_python

hacs_repository_data.py

Source:hacs_repository_data.py

1"""API Handler for hacs_repository_data"""2import sys3import homeassistant.helpers.config_validation as cv4import voluptuous as vol5from aiogithubapi import AIOGitHubAPIException6from homeassistant.components import websocket_api7from custom_components.hacs.helpers.classes.exceptions import HacsException8from custom_components.hacs.helpers.functions.logger import getLogger9from custom_components.hacs.helpers.functions.misc import extract_repository_from_url10from custom_components.hacs.helpers.functions.register_repository import (11    register_repository,12)13from custom_components.hacs.share import get_hacs14_LOGGER = getLogger()15@websocket_api.async_response16@websocket_api.websocket_command(17    {18        vol.Required("type"): "hacs/repository/data",19        vol.Optional("action"): cv.string,20        vol.Optional("repository"): cv.string,21        vol.Optional("data"): cv.string,22    }23)24async def hacs_repository_data(hass, connection, msg):25    """Handle get media player cover command."""26    hacs = get_hacs()27    repo_id = msg.get("repository")28    action = msg.get("action")29    data = msg.get("data")30    if repo_id is None:31        return32    if action == "add":33        repo_id = extract_repository_from_url(repo_id)34        if repo_id is None:35            return36        if repo_id in hacs.common.skip:37            hacs.common.skip.remove(repo_id)38        if not hacs.get_by_name(repo_id):39            try:40                registration = await register_repository(repo_id, data.lower())41                if registration is not None:42                    raise HacsException(registration)43            except (44                Exception,45                BaseException,46            ) as exception:  # pylint: disable=broad-except47                hass.bus.async_fire(48                    "hacs/error",49                    {50                        "action": "add_repository",51                        "exception": str(sys.exc_info()[0].__name__),52                        "message": str(exception),53                    },54                )55        else:56            hass.bus.async_fire(57                "hacs/error",58                {59                    "action": "add_repository",60                    "message": f"Repository '{repo_id}' exists in the store.",61                },62            )63        repository = hacs.get_by_name(repo_id)64    else:65        repository = hacs.get_by_id(repo_id)66    if repository is None:67        hass.bus.async_fire("hacs/repository", {})68        return69    _LOGGER.debug("Running %s for %s", action, repository.data.full_name)70    try:71        if action == "set_state":72            repository.state = data73        elif action == "set_version":74            repository.data.selected_tag = data75            await repository.update_repository()76            repository.state = None77        elif action == "install":78            was_installed = repository.data.installed79            repository.data.selected_tag = data80            await repository.update_repository()81            await repository.async_install()82            repository.state = None83            if not was_installed:84                hass.bus.async_fire("hacs/reload", {"force": True})85        elif action == "add":86            repository.state = None87        else:88            repository.state = None89            _LOGGER.error("WS action '%s' is not valid", action)90        message = None91    except AIOGitHubAPIException as exception:92        message = exception93    except AttributeError as exception:94        message = f"Could not use repository with ID {repo_id} ({exception})"95    except (Exception, BaseException) as exception:  # pylint: disable=broad-except96        message = exception97    if message is not None:98        _LOGGER.error(message)99        hass.bus.async_fire("hacs/error", {"message": str(message)})100    await hacs.data.async_write()...

service.py

Source:service.py

1from ..repository import Repository2from ..repository.mongo import MongoRepository3from .schema import KudoSchema4class Service(object):5 def __init__(self, user_id, repo_client=Repository(adapter=MongoRepository)):6   self.repo_client = repo_client7   self.user_id = user_id8   if not user_id:9     raise Exception("user id not provided")10 def find_all_kudos(self):11   kudos  = self.repo_client.find_all({'user_id': self.user_id})12   return [self.dump(kudo) for kudo in kudos]13 def find_kudo(self, repo_id):14   kudo = self.repo_client.find({'user_id': self.user_id, 'repo_id': repo_id})15   return self.dump(kudo)16 def create_kudo_for(self, githubRepo):17   self.repo_client.create(self.prepare_kudo(githubRepo))18   return self.dump(githubRepo.data)19 def update_kudo_with(self, repo_id, githubRepo):20   records_affected = self.repo_client.update({'user_id': self.user_id, 'repo_id': repo_id}, self.prepare_kudo(githubRepo))21   return records_affected > 022 def delete_kudo_for(self, repo_id):23   records_affected = self.repo_client.delete({'user_id': self.user_id, 'repo_id': repo_id})24   return records_affected > 025 def dump(self, data):26   return KudoSchema(exclude=['_id']).dump(data).data27 from ..repository import Repository28from ..repository.mongo import MongoRepository29from .schema import KudoSchema30class Service(object):31 def __init__(self, user_id, repo_client=Repository(adapter=MongoRepository)):32   self.repo_client = repo_client33   self.user_id = user_id34   if not user_id:35     raise Exception("user id not provided")36 def find_all_kudos(self):37   kudos  = self.repo_client.find_all({'user_id': self.user_id})38   return [self.dump(kudo) for kudo in kudos]39 def find_kudo(self, repo_id):40   kudo = self.repo_client.find({'user_id': self.user_id, 'repo_id': repo_id})41   return self.dump(kudo)42 def create_kudo_for(self, githubRepo):43   self.repo_client.create(self.prepare_kudo(githubRepo))44   return self.dump(githubRepo.data)45 def update_kudo_with(self, repo_id, githubRepo):46   records_affected = self.repo_client.update({'user_id': self.user_id, 'repo_id': repo_id}, self.prepare_kudo(githubRepo))47   return records_affected > 048 def delete_kudo_for(self, repo_id):49   records_affected = self.repo_client.delete({'user_id': self.user_id, 'repo_id': repo_id})50   return records_affected > 051 def dump(self, data):52   return KudoSchema(exclude=['_id']).dump(data).data53 def prepare_kudo(self, githubRepo):54   data = githubRepo.data55   data['user_id'] = self.user_id...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.