Python task

Task7/create_db.py

@@ -0,0 +1,7 @@
+from sqlmodel import SQLModel
+from database import engine
+
+def create_db_and_tables():
+    SQLModel.metadata.create_all(engine)
+if __name__ == "__main__":
+    create_db_and_tables()

Task7/database.py

@@ -0,0 +1,16 @@
+from sqlmodel import create_engine,Session
+from fastapi import Depends, FastAPI, HTTPException, Query
+from typing import Annotated
+
+
+
+DATABASE_URL = 'postgresql://postgres:postgres@localhost/fastapi'
+
+engine = create_engine(DATABASE_URL,echo=True)
+
+
+def get_session():
+   with Session(engine) as session:
+       yield session
+
+SessionDep = Annotated[Session, Depends(get_session)]

Task7/main.py

@@ -0,0 +1,147 @@
+from fastapi import FastAPI, Depends, HTTPException
+from sqlmodel import SQLModel, Session, select
+from database import engine, get_session
+from model import Project, Resource, ProjectResource
+from datetime import datetime
+
+app = FastAPI()
+
+
+# Initialize the database on startup
+@app.on_event("startup")
+def initialize_database():
+    SQLModel.metadata.create_all(engine)
+
+
+# Create a new project and assign a project manager
+@app.post("/projects/")
+def create_project(project: Project, session: Session = Depends(get_session)):
+    # Validate if the project manager exists
+    manager = session.exec(select(Resource).where(Resource.resource_id == project.project_manager_id)).first()
+    if not manager:
+        raise HTTPException(status_code=404, detail="Project manager does not exist")
+
+    # Add the new project to the database
+    session.add(project)
+    session.commit()
+    session.refresh(project)
+
+    # Assign the manager to the project in the ProjectResource table
+    manager_assignment = ProjectResource(
+        project_id=project.project_id,
+        resource_id=project.project_manager_id,
+        onboard=project.start_date
+    )
+    session.add(manager_assignment)
+    session.commit()
+
+    return project
+
+
+# Retrieve all projects
+@app.get("/projects/")
+def list_projects(session: Session = Depends(get_session)):
+    return session.exec(select(Project)).all()
+
+
+# Create a new resource (employee)
+@app.post("/resources/")
+def create_resource(resource: Resource, session: Session = Depends(get_session)):
+    session.add(resource)
+    session.commit()
+    session.refresh(resource)
+    return resource
+
+
+# Retrieve all resources
+@app.get("/resources/")
+def list_resources(session: Session = Depends(get_session)):
+    return session.exec(select(Resource)).all()
+
+
+# Assign a resource to a project
+@app.post("/project-resource/")
+def assign_resource_to_project(assignment: ProjectResource, session: Session = Depends(get_session)):
+    # Validate if the resource exists
+    resource = session.exec(select(Resource).where(Resource.resource_id == assignment.resource_id)).first()
+    if not resource:
+        raise HTTPException(status_code=404, detail="Resource not found")
+
+    # Assign resource to project
+    session.add(assignment)
+    session.commit()
+    session.refresh(assignment)
+    return assignment
+
+
+# Update project status and handle resource offboarding
+@app.put("/projects/{project_id}/status/")
+def update_project_status(project_id: int, status: str, session: Session = Depends(get_session)):
+    project = session.exec(select(Project).where(Project.project_id == project_id)).first()
+    if not project:
+        raise HTTPException(status_code=404, detail="Project not found")
+
+    project.status = status
+    if status == "completed":
+        project.end_date = datetime.utcnow()
+
+    session.commit()
+
+    if status in ["completed", "onhold"]:
+        # Offboard all resources assigned to this project
+        assigned_resources = session.exec(select(ProjectResource).where(ProjectResource.project_id == project_id)).all()
+        for resource in assigned_resources:
+            resource.offboard = datetime.utcnow()
+            session.add(resource)
+
+        session.commit()
+        update_bench_resources(session)
+
+    return {"message": f"Project status updated to {status}"}
+
+
+# Mark resources as "bench" if they are not assigned to any active projects
+def update_bench_resources(session: Session):
+    all_resources = session.exec(select(Resource)).all()
+    for resource in all_resources:
+        active_projects = session.exec(
+            select(ProjectResource).where(ProjectResource.resource_id == resource.resource_id,
+                                          ProjectResource.offboard == None)
+        ).all()
+
+        if not active_projects:
+            resource.status = "bench"
+            session.add(resource)
+
+    session.commit()
+
+
+# Retrieve all resources currently on the bench
+@app.get("/resources/bench/")
+def list_bench_resources(session: Session = Depends(get_session)):
+    return session.exec(select(Resource).where(Resource.status == "bench")).all()
+
+
+# Offboard a resource from a project
+@app.put("/project-resource/{record_id}/offboard/")
+def offboard_resource(record_id: int, session: Session = Depends(get_session)):
+    project_assignment = session.exec(select(ProjectResource).where(ProjectResource.record_id == record_id)).first()
+    if not project_assignment:
+        raise HTTPException(status_code=404, detail="Project resource entry not found")
+
+    project_assignment.offboard = datetime.utcnow()
+    session.commit()
+    return {"message": "Resource offboarded successfully"}
+
+
+# Retrieve all resources assigned to a specific project
+@app.get("/projects/{project_id}/resources/")
+def list_project_resources(project_id: int, session: Session = Depends(get_session)):
+    assigned_resources = session.exec(
+        select(Resource).join(ProjectResource).where(ProjectResource.project_id == project_id)
+    ).all()
+
+    if not assigned_resources:
+        raise HTTPException(status_code=404, detail="No resources found for this project")
+
+    return assigned_resources

Task7/model.py

@@ -0,0 +1,37 @@
+from sqlmodel import SQLModel, Field, Relationship
+from typing import Optional, List
+from datetime import datetime
+
+# class ProjectManager(SQLModel, table=True):
+#     id: Optional[int] = Field(default=None, primary_key=True)
+#     name: str
+#     email: str = Field(unique=True)
+#     projects: List["Project"] = Relationship(back_populates="manager")
+
+class ProjectResource(SQLModel, table=True):
+    record_id:Optional[int] = Field(default=None, primary_key=True)
+    project_id: Optional[int] = Field(default=None, foreign_key="project.project_id" )
+    resource_id: Optional[int] = Field(default=None, foreign_key="resource.resource_id")
+    onboard:datetime
+    offboard:Optional[datetime]=None
+
+class Project(SQLModel, table=True):
+    project_id: Optional[int] = Field(default=None, primary_key=True)
+    name: str
+    description: str
+    status: str = Field(default="ongoing")
+    deadline: Optional[datetime] = None
+    startdate:datetime
+    enddate:Optional[datetime]=None
+    project_manager_id: Optional[int] = Field(default=None, foreign_key="resource.resource_id")
+    # manager: Optional[ProjectManager] = Relationship(back_populates="projects")
+    resources: List["Resource"] = Relationship(back_populates="projects", link_model=ProjectResource)
+
+class Resource(SQLModel, table=True):
+    resource_id: Optional[int] = Field(default=None, primary_key=True)
+    name: str
+    role: str
+    status: str = Field(default="working")
+    email:str= Field(unique=True)
+    isActive:bool=Field(default=True)
+    projects: List[Project] = Relationship(back_populates="resources", link_model=ProjectResource)

task1/task1.py

@@ -0,0 +1,14 @@
+import requests
+import pip
+
+def install_package(version):
+    pip.main(['install', f'requests=={version}'])
+i_version = requests.__version__
+print(i_version)
+r_version = input("Enter the required requests version: ").strip()
+
+if i_version != r_version:
+    choice = input(f"Version mismatch! Required: {r_version}, Installed: {i_version}. Install required version? (y/n): ")
+    if choice.lower() == 'y':
+        install_package(r_version)
+        print(f"requests {r_version} installed successfully.")

task2/task2_1.py

@@ -0,0 +1,13 @@
+# Create a custom module with the function of converting array of string
+# to dictionary of string(key as sequence integer and value as string).
+# Import and use these functions from another python script to display
+# output in console
+
+
+def convert_to_dict(str):
+    dist={}
+    key=0
+    for i in str:
+        dist.update({key:i})
+        key+=1
+    return dist

task2/task2_2.py

@@ -0,0 +1,6 @@
+import task2_1
+
+str="Python comes with batteries included"
+
+result=task2_1.convert_to_dict(str)
+print(result)

task3/task3.py

@@ -0,0 +1,70 @@
+# Create a program that asks users for key-value pairs (name:age format)
+# and builds a dictionary. Then display the oldest and youngest person group
+# by age group as 18-35, 36-50, 51-65, and 66-90
+
+
+print("(1 for add your details , 2 for see the details,3 for exit )")
+status=True
+dist = {}
+
+while status:
+    choice=input("Enter your Choice: ")
+    name=""
+    age=""
+
+    def input_fn():
+        name=input("enter your name: ")
+        age=input("enter your age: ")
+        dist.update({name:int(age)})
+    def display():
+        youngest_group={}
+        second_group={}
+        third_group={}
+        oldest_group={}
+        for key,value in dist.items():
+            if(int(value)>=18 and int(value)<=35):
+                youngest_group.update({key:int(value)})
+            elif(int(value)>=36 and int(value)<=50):
+                second_group.update({key:int(value)})
+            elif(int(value)>=51 and int(value)<=65):
+                third_group.update({key:int(value)})
+            elif(int(value)>66 and int(value)<=90):
+                oldest_group.update({key:int(value)})
+        l1=youngest_group.values()
+        if(len(l1)>0):
+            min_first=min(l1)
+            max_first=max(l1)
+            print(f"in group 18-35 youngest is: {list(youngest_group.keys())[list(youngest_group.values()).index(min_first)]} with age: {min_first}")
+            print(f"in group 18-35 oldest is: {list(youngest_group.keys())[list(youngest_group.values()).index(max_first)]} with age: {max_first}")
+        l2=second_group.values()
+        if(len(l2)>0):
+            min_second=min(l2)
+            max_second=max(l2)
+            print(min_second)
+            print(f"in group 36-50 youngest is: {list(second_group.keys())[list(second_group.values()).index(min_second)]} with age: {min_second}")
+            print(f"in group 36-50 youngest is: {list(second_group.keys())[list(second_group.values()).index(max_second)]} with age: {max_second}")
+        l3 = third_group.values()
+        if(len(l3)>0):
+            min_third = min(l3)
+            max_third = max(l3)
+
+            print(f"in group 51-65 youngest is: {list(third_group.keys())[list(third_group.values()).index(min_third)]} with age: {min_third}")
+            print(f"in group 51-65 youngest is: {list(third_group.keys())[list(third_group.values()).index(max_third)]} with age: {max_third}")
+        l4 = oldest_group.values()
+        if(len(l4)>0):
+            min_fourth = min(l4)
+            max_fourth = max(l4)
+            print(f"in group 66-90 youngest is: {list(oldest_group.keys())[list(oldest_group.values()).index(min_fourth)]} with age: {min_fourth}")
+            print(f"in group 66-90 youngest is: {list(oldest_group.keys())[list(oldest_group.values()).index(max_fourth)]} with age: {max_fourth}")
+
+
+    if(choice=="1"):
+        input_fn()
+    elif(choice=="2"):
+        display()
+    elif(choice=="3"):
+        status=False
+    else:
+        print("enter valid no")
+        exit()
+

task4/task4.py

@@ -0,0 +1,30 @@
+# From a list of any 1000 numbers, create three new data collections:
+# even numbers, numbers multiplier of 17, and numbers greater than a user-specified
+# input number
+
+
+import random
+l=[]
+listofno=[10,100,1000,10000,100000]
+for i in range(1000):
+    l.append(int(random.random()*random.choice(listofno)))
+print(l)
+even_no=set()
+no_multiplier_17=set()
+no_grater_choice=set()
+no=int(input("enter your favourite no: "))
+for i in l:
+    if(i%2==0):
+        even_no.add(i)
+    if(i%17==0):
+        no_multiplier_17.add(i)
+    if(i>no):
+        no_grater_choice.add(i)
+print("Even no: ",end=' ')
+print(even_no)
+print("numbers multiplier of 17: ",end=' ')
+print(no_multiplier_17)
+print("numbers greater than a user-specified input number: ",end=' ')
+print(no_grater_choice)
+
+

task5/Add_Details.py

@@ -0,0 +1,19 @@
+
+
+
+def addDetails():
+    id=input("enter your id: ")
+    name = input("enter your name: ")
+    luckyNo = input("enter your lucky no: ")
+    hobbies = []
+    ho_no = input("how many hobbies do you have: ")
+    print("enter your hobbies: ", end=" ")
+    for i in range(int(ho_no)):
+        hobbies.append(input())
+
+    print("enter book you read with author: ", end=" ")
+    book = []
+    book.append(input())
+    book.append(input())
+    d = {"id": id, "name": name, "lucky_no": luckyNo, "hobbies": hobbies, "book": book}
+    return d