[docs]classBaseGitHubLoader(BaseLoader,BaseModel,ABC):"""Load `GitHub` repository Issues."""repo:str"""Name of repository"""access_token:str"""Personal access token - see https://github.com/settings/tokens?type=beta"""github_api_url:str="https://api.github.com""""URL of GitHub API"""@model_validator(mode="before")@classmethoddefvalidate_environment(cls,values:Dict)->Any:"""Validate that access token exists in environment."""values["access_token"]=get_from_dict_or_env(values,"access_token","GITHUB_PERSONAL_ACCESS_TOKEN")returnvalues@propertydefheaders(self)->Dict[str,str]:return{"Accept":"application/vnd.github+json","Authorization":f"Bearer {self.access_token}",}
[docs]classGitHubIssuesLoader(BaseGitHubLoader):"""Load issues of a GitHub repository."""include_prs:bool=True"""If True include Pull Requests in results, otherwise ignore them."""milestone:Union[int,Literal["*","none"],None]=None"""If integer is passed, it should be a milestone's number field. If the string '*' is passed, issues with any milestone are accepted. If the string 'none' is passed, issues without milestones are returned. """state:Optional[Literal["open","closed","all"]]=None"""Filter on issue state. Can be one of: 'open', 'closed', 'all'."""assignee:Optional[str]=None"""Filter on assigned user. Pass 'none' for no user and '*' for any user."""creator:Optional[str]=None"""Filter on the user that created the issue."""mentioned:Optional[str]=None"""Filter on a user that's mentioned in the issue."""labels:Optional[List[str]]=None"""Label names to filter one. Example: bug,ui,@high."""sort:Optional[Literal["created","updated","comments"]]=None"""What to sort results by. Can be one of: 'created', 'updated', 'comments'. Default is 'created'."""direction:Optional[Literal["asc","desc"]]=None"""The direction to sort the results by. Can be one of: 'asc', 'desc'."""since:Optional[str]=None"""Only show notifications updated after the given time. This is a timestamp in ISO 8601 format: YYYY-MM-DDTHH:MM:SSZ."""page:Optional[int]=None"""The page number for paginated results. Defaults to 1 in the GitHub API."""per_page:Optional[int]=None"""Number of items per page. Defaults to 30 in the GitHub API."""@field_validator("since")@classmethoddefvalidate_since(cls,v:Optional[str])->Optional[str]:ifv:try:datetime.strptime(v,"%Y-%m-%dT%H:%M:%SZ")exceptValueError:raiseValueError("Invalid value for 'since'. Expected a date string in "f"YYYY-MM-DDTHH:MM:SSZ format. Received: {v}")returnv
[docs]deflazy_load(self)->Iterator[Document]:""" Get issues of a GitHub repository. Returns: A list of Documents with attributes: - page_content - metadata - url - title - creator - created_at - last_update_time - closed_time - number of comments - state - labels - assignee - assignees - milestone - locked - number - is_pull_request """url:Optional[str]=self.urlwhileurl:response=requests.get(url,headers=self.headers)response.raise_for_status()issues=response.json()forissueinissues:doc=self.parse_issue(issue)ifnotself.include_prsanddoc.metadata["is_pull_request"]:continueyielddocif(response.linksandresponse.links.get("next")and(notself.pageandnotself.per_page)):url=response.links["next"]["url"]else:url=None
[docs]defparse_issue(self,issue:dict)->Document:"""Create Document objects from a list of GitHub issues."""metadata={"url":issue["html_url"],"title":issue["title"],"creator":issue["user"]["login"],"created_at":issue["created_at"],"comments":issue["comments"],"state":issue["state"],"labels":[label["name"]forlabelinissue["labels"]],"assignee":issue["assignee"]["login"]ifissue["assignee"]elseNone,"milestone":issue["milestone"]["title"]ifissue["milestone"]elseNone,"locked":issue["locked"],"number":issue["number"],"is_pull_request":"pull_request"inissue,}content=issue["body"]ifissue["body"]isnotNoneelse""returnDocument(page_content=content,metadata=metadata)
@propertydefquery_params(self)->str:"""Create query parameters for GitHub API."""labels=",".join(self.labels)ifself.labelselseself.labelsquery_params_dict={"milestone":self.milestone,"state":self.state,"assignee":self.assignee,"creator":self.creator,"mentioned":self.mentioned,"labels":labels,"sort":self.sort,"direction":self.direction,"since":self.since,"page":self.page,"per_page":self.per_page,}query_params_list=[f"{k}={v}"fork,vinquery_params_dict.items()ifvisnotNone]query_params="&".join(query_params_list)returnquery_params@propertydefurl(self)->str:"""Create URL for GitHub API."""returnf"{self.github_api_url}/repos/{self.repo}/issues?{self.query_params}"