Skip to content

Commit

Permalink
Merge pull request #3 from kubetoolsca/coding-standards-0.1.1
Browse files Browse the repository at this point in the history
Code Improvements Complete
  • Loading branch information
abhimazu authored Jun 11, 2024
2 parents 1b96a2d + 25ec423 commit f77ccd0
Show file tree
Hide file tree
Showing 6 changed files with 1,763 additions and 591 deletions.
184 changes: 147 additions & 37 deletions krs/krs.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,66 +4,176 @@
from krs.main import KrsMain
from krs.utils.constants import KRSSTATE_PICKLE_FILEPATH, KRS_DATA_DIRECTORY

app = typer.Typer()
krs = KrsMain()
app = typer.Typer() # Main typer app
krs = KrsMain() # Main class object

def check_initialized():
if not os.path.exists(KRSSTATE_PICKLE_FILEPATH):
typer.echo("KRS is not initialized. Please run 'krs init' first.")
raise typer.Exit()

if not os.path.exists(KRS_DATA_DIRECTORY):
os.mkdir(KRS_DATA_DIRECTORY)
def check_initialized() -> None:

"""
Checks if KRS is initialized or not.
"""
try:
if not os.path.exists(KRSSTATE_PICKLE_FILEPATH):
typer.echo("KRS is not initialized. Please run 'krs init' first.")
raise typer.Exit()
except Exception as e:
typer.echo("Error: ", e)
raise typer.Abort()
except KeyboardInterrupt:
typer.echo("\nExiting...")
raise typer.Abort()
except typer.Exit:
raise typer.Abort()
except typer.Abort:
raise typer.Abort()
except:
typer.echo("An error occured. Please try again.")
raise typer.Abort()

@app.command()
def init():
try:
if not os.path.exists(KRS_DATA_DIRECTORY): # Create data directory if not exists
os.mkdir(KRS_DATA_DIRECTORY)
except Exception as e:
typer.echo("Error: ", e)
raise typer.Abort()
except KeyboardInterrupt as e:
typer.echo("\nInterruption: ", e)
raise typer.Abort()
except typer.Exit as e:
typer.echo("\nExiting: ", e)
raise typer.Abort()
except typer.Abort as e:
typer.echo("\nAborting: ", e)
raise typer.Abort()
except:
typer.echo("An error occured. Please try again.")
raise typer.Abort()


@app.command() # Command to initialize the services
def init() -> None: # Function to initialize the services
"""
Initializes the services and loads the scanner.
"""
krs.initialize()
typer.echo("Services initialized and scanner loaded.")
try:
krs.initialize()
typer.echo("Services initialized and scanner loaded.")
except Exception as e:
typer.echo("Error: ", e)
raise typer.Abort()
except KeyboardInterrupt as e:
typer.echo("\nInterruption: ", e)
raise typer.Abort()
except typer.Exit as e:
typer.echo("\nExiting: ", e)
raise typer.Abort()
except typer.Abort as e:
typer.echo("\nAborting: ", e)
raise typer.Abort()
except:
typer.echo("An error occured. Please try again.")
raise typer.Abort()

# Command to scan the cluster
@app.command()
def scan():
def scan() -> None:
"""
Scans the cluster and extracts a list of tools that are currently used.
"""
check_initialized()
krs.scan_cluster()

try:
check_initialized()
krs.scan_cluster()
except Exception as e:
typer.echo("Error: ", e)
raise typer.Abort()
except KeyboardInterrupt as e:
typer.echo("\nInterruption: ", e)
raise typer.Abort()
except typer.Exit as e:
typer.echo("\nExiting: ", e)
raise typer.Abort()
except typer.Abort as e:
typer.echo("\nAborting: ", e)
raise typer.Abort()
except:
typer.echo("An error occured. Please try again.")
raise typer.Abort()

# Command to list all the namespaces
@app.command()
def namespaces():
def namespaces() -> None:
"""
Lists all the namespaces.
"""
check_initialized()
namespaces = krs.list_namespaces()
typer.echo("Namespaces in your cluster are: \n")
for i, namespace in enumerate(namespaces):
typer.echo(str(i+1)+ ". "+ namespace)

try:
check_initialized()
namespaces = krs.list_namespaces()
typer.echo("Namespaces in your cluster are: \n")
for i, namespace in enumerate(namespaces):
typer.echo(str(i+1)+ ". "+ namespace)
except Exception as e:
typer.echo("Error: ", e)
raise typer.Abort()
except KeyboardInterrupt as e:
typer.echo("\nInterruption: ", e)
raise typer.Abort()
except typer.Exit as e:
typer.echo("\nExiting: ", e)
raise typer.Abort()
except typer.Abort as e:
typer.echo("\nAborting: ", e)
raise typer.Abort()
except:
typer.echo("An error occured. Please try again.")
raise typer.Abort()

# Command to list all the pods
@app.command()
def pods(namespace: str = typer.Option(None, help="Specify namespace to list pods from")):
def pods(namespace: str = typer.Option(None, help="Specify namespace to list pods from")) -> None:

"""
Lists all the pods with namespaces, or lists pods under a specified namespace.
"""
check_initialized()
if namespace:
pods = krs.list_pods(namespace)
if pods == 'wrong namespace name':
typer.echo(f"\nWrong namespace name entered, try again!\n")
raise typer.Abort()
typer.echo(f"\nPods in namespace '{namespace}': \n")
else:
pods = krs.list_pods_all()
typer.echo("\nAll pods in the cluster: \n")
for i, pod in enumerate(pods):
typer.echo(str(i+1)+ '. '+ pod)
Args:
namespace: str: Namespace name to list pods from.
Returns:
None
"""
try:
check_initialized()
if namespace:
pods = krs.list_pods(namespace)
if pods == 'wrong namespace name':
typer.echo(f"\nWrong namespace name entered, try again!\n")
raise typer.Abort()
typer.echo(f"\nPods in namespace '{namespace}': \n")
else:
pods = krs.list_pods_all()
typer.echo("\nAll pods in the cluster: \n")

for i, pod in enumerate(pods):
typer.echo(str(i+1)+ '. '+ pod)
except Exception as e:
typer.echo("Error: ", e)
raise typer.Abort()
except KeyboardInterrupt as e:
typer.echo("\nInterruption: ", e)
raise typer.Abort()
except typer.Exit as e:
typer.echo("\nExiting: ", e)
raise typer.Abort()
except typer.Abort as e:
typer.echo("\nAborting: ", e)
raise typer.Abort()
except:
typer.echo("An error occured. Please try again.")
raise typer.Abort()


@app.command()
def recommend():
def recommend(): # Command to recommend tools
"""
Generates a table of recommended tools from our ranking database and their CNCF project status.
"""
Expand Down
Loading

0 comments on commit f77ccd0

Please sign in to comment.