first commit

Exceptions.py

+class Error(Exception):
+    pass
+
+
+class UniqueCredentialError(Error):
+    pass
+
+
+class ValueTooLargeError(Exception):
+    """Raised when the input value is too large"""
+    pass

main.py

-print('hello')
\ No newline at end of file
+import fastapi
+import pymongo.errors
+from fastapi import FastAPI
+from typing import Optional
+from Exceptions import UniqueCredentialError
+from test import myCollection, LaunchCredential, User, DockerExecutorEngine
+
+app = FastAPI()
+
+
+@app.post('/credentials', tags=["Enter Credentials"])
+async def add_new_credentials(user: User):
+    try:
+        _id = myCollection.insert_one(dict(user))
+        if _id:
+            return {"status": "sucess"}
+
+    # except pymongo.errors.DuplicateKeyError as d:
+    #     print(d)
+    #     return {"message": f"credential name already exist, try again! {d}"}
+    except Exception as e:
+        return {"status": "failed", "message": f"credential name already exist, try again! {e.__class__}"}, 500
+
+
+@app.get("/credentials", tags=["List of Credentials"])
+def employee_details(
+        category: str = fastapi.Query(None, description="Enter Type of credentials or empty field will display all")):
+    if category:
+        return myCollection.find_one({"type": category}, {"_id": 0, "parameters": {"password": 0}})
+    return list(myCollection.find({}, {"_id": 0, "parameters": {"password": 0}}))
+
+
+"""class LaunchCredential(BaseModel): credential_name: Optional[str] = "satyam-creds" image: str variable_parameters: 
+Optional[dict] = {"ports":{"8002":"5000"},"environment":{"key": "value"},"volumes":"(local_dir, "/webroot")"} 
+"""
+
+
+@app.post('/launchContainer', tags=["Enter details to launch container"])
+def add_new_docker_credentials(user: LaunchCredential):
+    docker_obj = DockerExecutorEngine(user.credential_name)
+    try:
+        return str(docker_obj.launch_container(user.image, **user.variable_parameters))
+    except Exception:
+        return {"status": "failed", "message": "Invalid input data, try again!"}, 500
+
+
+# @app.post('/launchContainer', tags=["Enter details to launch container"])
+# def add_new_docker_credentials(credentail_name: Optional[str] = None, image: str = None, data: Optional[dict] = {}):
+#
+#     docker_obj = DockerExecutorEngine(credentail_name)
+#     return str(docker_obj.launch_container(image, **data))
+
+
+
+
+
+@app.get("/status", tags=["Check the status of container"])
+def status_of_launched_containers(Container_id: str = fastapi.Query(None, description="Enter conteiner ID or empty "
+                                                                                      "field will display all "
+                                                                                      "containers ")):
+    docker_obj = DockerExecutorEngine()
+    if Container_id:
+        return docker_obj.container_status(Container_id)
+    ll = (docker_obj.all_container_status())
+    result = []
+    for x in ll:
+        result.append(x.id)
+    return {"list_of_containers": result}
+
+
+
+
+
+@app.get("/stop", tags=["Terminate container"])
+def remove_containers(Container_id: str):
+    docker_obj = DockerExecutorEngine()
+    return docker_obj.terminate_container(Container_id)

requirements.txt

 anyio==3.5.0

test.py

+import docker
+import pymongo
+from typing import Optional
+from pydantic import BaseModel
+
+client = pymongo.MongoClient("mongodb://localhost:27017/")
+# Access database
+myDatabase = client['sankar_db']
+# Access collection of the database
+myCollection = myDatabase['docker_credentials']
+
+
+class User(BaseModel):
+    credential_name: str
+    parameters: dict = {"Username": "osr2000", "password": "oO@096689", "email": "obbintisankarrao@gmail.com"}
+    type: str = "docker/azure-blob"
+
+
+class LaunchCredential(BaseModel):
+    credential_name: Optional[str] = "satyam-creds"
+    image: str
+    variable_parameters: Optional[dict] = {"ports": {"8002": "5000"}, "environment": {"key": "value"}}
+
+
+class DockerExecutorEngine:
+    def __init__(self, credential_name=""):
+        """
+            credential-name:
+            image-name(name:tag):
+            env-variable:
+            volume_mounts(optional):
+            port-to-exposed:
+            name-of-launched-container:
+            tags(optional):
+            network(optional):
+            command(optional):
+        """
+        self.credential_name = credential_name
+        self.client = docker.DockerClient.from_env()
+        if self.credential_name:
+            self.login_docker_account()
+
+    def login_docker_account(self):
+        ob = myCollection.find_one({"credential_name": self.credential_name, "type": "docker"},
+                                   {"_id": 0, "credential_name": 0, "parameters": {"registry_name": 0}, "type": 0,
+                                    "meta": 0})
+        if not ob:
+            raise ValueError("Invalid credentials")
+
+        self.client.login(username=str(ob['parameters']['username']), password=str(ob['parameters']['password']),
+                          email=str(ob['parameters']['email']),
+                          registry='https://index.docker.io/v1')
+
+
+    def launch_container(self, image, **kwargs):
+        return self.client.containers.run(image=image, detach=True, **kwargs)
+
+
+
+    def container_status(self, container_id):
+        a = self.container_details(container_id)
+        b = self.client.containers.get(container_id).status
+        data = [container_id, a['Name']]
+        resp = {"data": data, "status": b, "message": "Container launched successfully"}
+        return resp
+
+    def all_container_status(self):
+        return self.client.containers.list()
+
+    def container_details(self, container_id):
+        return self.client.containers.get(container_id).attrs
+
+    def stop_container(self, container_id):
+        return self.client.containers.get(container_id).stop()
+
+    def terminate_container(self, container_id):
+        self.stop_container(container_id)
+        return self.client.containers.get(container_id).remove()
+
+
+if __name__ == "__main__":
+    docker_obj = DockerExecutorEngine(credential_name="satyam-creds")
+    res = docker_obj.login_docker_account()
+    print(res)
+
+    status = docker_obj.launch_container(image="redis:latest", detach=True, ports={800: 800})
+    # status = docker_obj.launch_container(image="redis:latest", command=["/bin/sh", "-c", 'echo 1 && echo 2'],
+    #                                      ports={10080: 8000},
+    #                                      volumes={'/PycharmProjects/newdocker/': {'bind': '/mnt/vol2', 'mode': 'rw'},
+    #                                               '/var/www': {'bind': '/mnt/vol1', 'mode': 'ro'}})
+    print(status)
+
+    # print(docker_obj.stop_container("25d59d85223b"))
+    # print(docker_obj.terminate_container('c823b13533'))

trail.py

+import docker
+from typing import Optional
+
+client = docker.from_env()
+xx=client.containers.get('e57c61c057e1').attrs
+
+print(type(xx['Name']))
+# res = client.login(username='osr2000', password='oO@096689', email='obbinti.rao@knowledgelens.com',
+#                    registry='https://index.docker.io/v1')
+
+# def testing_method(image: str, myDict: Optional[dict]):
+#     if myDict:
+#         s = client.containers.run(image,detach=True, **myDict)
+#         print(s)
+#     s = client.containers.run(image,detach=True)
+#     print(s)
+#
+# testing_method('redis',{})
+
+
+# def testing_method(image: str, myDict: Optional[dict]):
+#     s = client.containers.run(image,detach=True, **myDict)
+#     print(s)
+# testing_method('redis',{})
+# # testing_method('redis')
+
+
+
+
+# print
+# s = client.containers.run(image='redis', detach=True)
+# print(s)c823b13533
+# print(client.containers.get('c823b13533').status)
+