Skip to content

Commit

Permalink
correct the format
Browse files Browse the repository at this point in the history
  • Loading branch information
aghadge committed Jan 27, 2025
1 parent 608ddb1 commit 9d897e3
Show file tree
Hide file tree
Showing 4 changed files with 89 additions and 112 deletions.
17 changes: 17 additions & 0 deletions docs/docs/integrations/tools/jenkins.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -173,6 +173,23 @@
"source": [
"tools[0].invoke({\"job\": \"job01\", \"action\": \"delete\"})"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Chaining\n",
"Will comming soon..\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## API reference\n",
"\n",
"For detailed documentation [API reference](https://python.langchain.com/docs/integrations/tools/jenkins/)"
]
}
],
"metadata": {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -2,4 +2,4 @@

from langchain_community.tools.jenkins.tool import JenkinsJobRun

__all__ = ["JenkinsJobRun"]
__all__ = ["JenkinsJobRun"]
128 changes: 55 additions & 73 deletions libs/community/langchain_community/tools/jenkins/tool.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Tool for the Jenkins API"""

from typing import Optional, Type
from typing import Any, Optional, Type

from langchain_core.callbacks import CallbackManagerForToolRun
from langchain_core.tools import BaseTool
Expand All @@ -11,92 +11,77 @@

class JenkinsSchema(BaseModel):
"""Input for the Jenkins tool.
Instantiate:
.. code-block:: python
from tools.jenkins.tool import JenkinsJobRun
from tools.jenkins.utility import JenkinsAPIWrapper
tools = [JenkinsJobRun(
api_wrapper=JenkinsAPIWrapper(
jenkins_server="https://jenkinsserver.com",
username="admin",
password=os.environ["PASSWORD"]
)
)]
Invoke directly with args:
.. code-block:: python
# delete jenkins job
tools[0].invoke({'job': "job01", "action": "delete"})
# create jenkins job
jenkins_job_content = ""
src_file = "job1.xml"
with open(src_file) as fread:
jenkins_job_content = fread.read()
tools[0].invoke({'job': "job01", "config_xml": jenkins_job_content,
"action": "create"})
# run the jenkins job
tools[0].invoke({'job': "job01", "parameters": {}, "action": "run"})
# get jenkins job info by passing job number
resp = tools[0].invoke({'job': "job01", "number": 1,
"action": "status"})
if not resp["inProgress"]:
print(resp["result"])
Instantiate:
.. code-block:: python
from tools.jenkins.tool import JenkinsJobRun
from tools.jenkins.utility import JenkinsAPIWrapper
tools = [JenkinsJobRun(
api_wrapper=JenkinsAPIWrapper(
jenkins_server="https://jenkinsserver.com",
username="admin",
password=os.environ["PASSWORD"]
)
)]
Invoke directly with args:
.. code-block:: python
# delete jenkins job
tools[0].invoke({'job': "job01", "action": "delete"})
# create jenkins job
jenkins_job_content = ""
src_file = "job1.xml"
with open(src_file) as fread:
jenkins_job_content = fread.read()
tools[0].invoke({'job': "job01", "config_xml": jenkins_job_content,
"action": "create"})
# run the jenkins job
tools[0].invoke({'job': "job01", "parameters": {}, "action": "run"})
# get jenkins job info by passing job number
resp = tools[0].invoke({'job': "job01", "number": 1,
"action": "status"})
if not resp["inProgress"]:
print(resp["result"])
"""

job: str = Field(
description="name of the job"
)
action: str = Field(
description="action of the job, like, create, run, delete"
)
number: int = Field(
default=1,
description="job number"
)
config_xml: str = Field(
default='',
description="job xml content"
)
parameters: dict = Field(
default={},
description="job parameters"
)
job: str = Field(description="name of the job")
action: str = Field(description="action of the job, like, create, run, delete")
number: int = Field(default=1, description="job number")
config_xml: str = Field(default="", description="job xml content")
parameters: dict = Field(default={}, description="job parameters")


class JenkinsJobRun(BaseTool): # type: ignore[override, override]
"""Tool that execute the job"""

name: str = "jenkins"
description: str = (
"""A tool that is used to create, trigger and delete Jenkins jobs with,
specified parameters."""
)
description: str = """A tool that is used to create,
trigger and delete Jenkins jobs with, specified parameters."""
api_wrapper: JenkinsAPIWrapper = Field(default_factory=JenkinsAPIWrapper) # type: ignore[arg-type]
args_schema: Type[BaseModel] = JenkinsSchema


def _run(
self,
job: str,
action: str,
number: Optional[int] = 1,
config_xml: Optional[str] = "",
parameters: Optional[dict] = {},
number: int = 1,
config_xml: str = "",
parameters: dict = {},
run_manager: Optional[CallbackManagerForToolRun] = None,
) -> any:
) -> Any:
"""Use the tool."""
if action == "create":
self.api_wrapper.create_job(
return self.api_wrapper.create_job(
job=job,
config_xml=config_xml,
)
Expand All @@ -110,9 +95,6 @@ def _run(
job=job,
)
elif action == "status":
return self.api_wrapper.status_job(
job=job,
number=number
)
return self.api_wrapper.status_job(job=job, number=number)
else:
raise ValueError("'action' not matched")
54 changes: 16 additions & 38 deletions libs/community/langchain_community/utilities/jenkins.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
"""Wrapper for the Jenkins API"""

import time
from typing import Any, Dict, Optional

Expand All @@ -10,9 +11,8 @@ class JenkinsAPIWrapper(BaseModel):
"""Wrapper for Jenkins API
To use, set the environment variables ``JENKINS_SERVER``,
``USERNAME`` and ``PASSWORD``. OR those input can supplay by
API parameter.
"""
``USERNAME`` and ``PASSWORD``. OR those input can supplay by API parameter."""

jenkins_client: Any

jenkins_server: Optional[str]
Expand All @@ -28,14 +28,10 @@ def validate_environment(cls, values: Dict) -> Any:
)
values["jenkins_server"] = jenkins_server

username = get_from_dict_or_env(
values, "username", "USERNAME"
)
username = get_from_dict_or_env(values, "username", "USERNAME")
values["username"] = username

password = get_from_dict_or_env(
values, "password", "PASSWORD"
)
password = get_from_dict_or_env(values, "password", "PASSWORD")
values["password"] = password

try:
Expand All @@ -46,47 +42,29 @@ def validate_environment(cls, values: Dict) -> Any:
please install it with pip install python-jenkins"""
)

jenkins_client = Jenkins(jenkins_server,
username=username,
password=password)
jenkins_client = Jenkins(jenkins_server, username=username, password=password)
values["jenkins_client"] = jenkins_client

return values

def delete_job(
self,
job: str
) -> None:
def delete_job(self, job: str) -> Any:
try:
self.jenkins_client.delete_job(job)
return self.jenkins_client.delete_job(job)
except Exception:
pass

def create_job(
self,
job: str,
config_xml: str
) -> None:
self.jenkins_client.create_job(
job, config_xml)

def run_job(
self,
job: str,
parameters: dict[str]
) -> int:
next_build_number = self.jenkins_client.get_job_info(job)['nextBuildNumber']
def create_job(self, job: str, config_xml: str) -> Any:
return self.jenkins_client.create_job(job, config_xml)

def run_job(self, job: str, parameters: dict) -> int:
next_build_number = self.jenkins_client.get_job_info(job)["nextBuildNumber"]
self.jenkins_client.build_job(job, parameters=parameters)
return next_build_number

def status_job(
self,
job: str,
number: int
) -> int:
def status_job(self, job: str, number: int) -> Any:
from jenkins import JenkinsException, NotFoundException

try:
return self.jenkins_client.get_build_info(job, number)
return self.jenkins_client.get_build_info(job, number)
except (NotFoundException, JenkinsException):
time.sleep(5)
return "Not Started.."

0 comments on commit 9d897e3

Please sign in to comment.