Kaadan commited on
Commit
a720bd6
·
1 Parent(s): 4920841

adding my applications endpoint

Browse files
backend/alembic/versions/f9f1aa7380ab_add_created_at_and_updated_at_columns_.py ADDED
@@ -0,0 +1,43 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """Add created_at and updated_at columns to applications table
2
+
3
+ Revision ID: f9f1aa7380ab
4
+ Revises: 9facd9b60600
5
+ Create Date: 2026-02-05 19:49:38.970805
6
+
7
+ """
8
+ from typing import Sequence, Union
9
+
10
+ from alembic import op
11
+ import sqlalchemy as sa
12
+
13
+
14
+ # revision identifiers, used by Alembic.
15
+ revision: str = 'f9f1aa7380ab'
16
+ down_revision: Union[str, Sequence[str], None] = '9facd9b60600'
17
+ branch_labels: Union[str, Sequence[str], None] = None
18
+ depends_on: Union[str, Sequence[str], None] = None
19
+
20
+
21
+ def upgrade() -> None:
22
+ """Upgrade schema."""
23
+ # ### commands auto generated by Alembic - please adjust! ###
24
+ # Add columns without default values first
25
+ op.add_column('applications', sa.Column('created_at', sa.DateTime(timezone=True), nullable=True))
26
+ op.add_column('applications', sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True))
27
+
28
+ # Update existing rows to set created_at to current timestamp
29
+ # This is needed because SQLite doesn't allow adding columns with non-constant defaults
30
+ connection = op.get_bind()
31
+ connection.execute(sa.text("UPDATE applications SET created_at = CURRENT_TIMESTAMP WHERE created_at IS NULL"))
32
+
33
+ # Now alter the column to add the default value
34
+ # Note: SQLite doesn't support altering column defaults easily, so we'll leave it as nullable
35
+ # ### end Alembic commands ###
36
+
37
+
38
+ def downgrade() -> None:
39
+ """Downgrade schema."""
40
+ # ### commands auto generated by Alembic - please adjust! ###
41
+ op.drop_column('applications', 'updated_at')
42
+ op.drop_column('applications', 'created_at')
43
+ # ### end Alembic commands ###
backend/api/application_routes.py CHANGED
@@ -4,9 +4,10 @@ from typing import List
4
  import json
5
 
6
  from database.database import get_db
7
- from schemas import ApplicationCreate, ApplicationUpdate, ApplicationResponse, ApplicationListResponse, ApplicationDetailedResponse, ApplicationDetailedListResponse
8
- from services import create_application, get_application, get_applications_by_job_and_assessment, calculate_application_score
9
  from services.assessment_service import get_assessment
 
10
  from utils.dependencies import get_current_user
11
  from models.user import User
12
  from logging_config import get_logger
@@ -78,14 +79,7 @@ def get_applications_list(jid: str, aid: str, page: int = 1, limit: int = 10, db
78
  def get_application_detail(jid: str, aid: str, id: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
79
  """Get detailed application information including answers"""
80
  logger.info(f"Retrieving application detail for job ID: {jid}, assessment ID: {aid}, application ID: {id} by user: {current_user.id}")
81
- # Only HR users can view application details
82
- if current_user.role != "hr":
83
- logger.warning(f"Unauthorized attempt to view application detail by user: {current_user.id} with role: {current_user.role}")
84
- raise HTTPException(
85
- status_code=status.HTTP_403_FORBIDDEN,
86
- detail="Only HR users can view application details"
87
- )
88
-
89
  # Get the application
90
  application = get_application(db, id)
91
  if not application or application.job_id != jid or application.assessment_id != aid:
@@ -95,6 +89,14 @@ def get_application_detail(jid: str, aid: str, id: str, db: Session = Depends(ge
95
  detail="Application not found for this job and assessment"
96
  )
97
 
 
 
 
 
 
 
 
 
98
  # Get the assessment to retrieve the passing score
99
  assessment = get_assessment(db, aid)
100
  if not assessment:
@@ -227,4 +229,55 @@ def create_new_application(jid: str, aid: str, application: ApplicationCreate, d
227
 
228
  db_application = create_application(db, application)
229
  logger.info(f"Successfully created application with ID: {db_application.id} for job ID: {jid}, assessment ID: {aid}")
230
- return {"id": db_application.id}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
4
  import json
5
 
6
  from database.database import get_db
7
+ from schemas import ApplicationCreate, ApplicationUpdate, ApplicationResponse, ApplicationListResponse, ApplicationDetailedResponse, ApplicationDetailedListResponse, MyApplicationsListResponse, MyApplicationResponse, MyApplicationsJob, MyApplicationsAssessment
8
+ from services import create_application, get_application, get_applications_by_job_and_assessment, calculate_application_score, get_applications_by_user
9
  from services.assessment_service import get_assessment
10
+ from services.job_service import get_job
11
  from utils.dependencies import get_current_user
12
  from models.user import User
13
  from logging_config import get_logger
 
79
  def get_application_detail(jid: str, aid: str, id: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
80
  """Get detailed application information including answers"""
81
  logger.info(f"Retrieving application detail for job ID: {jid}, assessment ID: {aid}, application ID: {id} by user: {current_user.id}")
82
+
 
 
 
 
 
 
 
83
  # Get the application
84
  application = get_application(db, id)
85
  if not application or application.job_id != jid or application.assessment_id != aid:
 
89
  detail="Application not found for this job and assessment"
90
  )
91
 
92
+ # Authorization: Allow HR users or the applicant who owns the application
93
+ if current_user.role != "hr" and current_user.id != application.user_id:
94
+ logger.warning(f"Unauthorized attempt to view application detail by user: {current_user.id} with role: {current_user.role}")
95
+ raise HTTPException(
96
+ status_code=status.HTTP_403_FORBIDDEN,
97
+ detail="Only HR users or the applicant who submitted the application can view application details"
98
+ )
99
+
100
  # Get the assessment to retrieve the passing score
101
  assessment = get_assessment(db, aid)
102
  if not assessment:
 
229
 
230
  db_application = create_application(db, application)
231
  logger.info(f"Successfully created application with ID: {db_application.id} for job ID: {jid}, assessment ID: {aid}")
232
+ return {"id": db_application.id}
233
+
234
+
235
+ @router.get("/my-applications", response_model=MyApplicationsListResponse)
236
+ def get_my_applications(page: int = 1, limit: int = 10, db: Session = Depends(get_db), current_user: User = Depends(get_current_user)):
237
+ """Get list of applications for the current logged-in user"""
238
+ logger.info(f"Retrieving applications for user ID: {current_user.id}, page: {page}, limit: {limit}")
239
+
240
+ skip = (page - 1) * limit
241
+ applications = get_applications_by_user(db, current_user.id, skip=skip, limit=limit)
242
+
243
+ # Calculate total count
244
+ total = len(get_applications_by_user(db, current_user.id, skip=0, limit=1000)) # Simplified for demo
245
+
246
+ # Create responses with job and assessment details
247
+ application_responses = []
248
+ for application in applications:
249
+ # Calculate score
250
+ score = calculate_application_score(db, application.id)
251
+
252
+ # Get assessment to retrieve passing score
253
+ assessment = get_assessment(db, application.assessment_id)
254
+
255
+ # Get job details
256
+ job = get_job(db, application.job_id)
257
+
258
+ # Create response object that matches technical requirements exactly
259
+ application_response = MyApplicationResponse(
260
+ id=application.id,
261
+ job=MyApplicationsJob(
262
+ id=job.id if job else "",
263
+ title=job.title if job else "",
264
+ seniority=job.seniority if job else "",
265
+ description=job.description if job else ""
266
+ ) if job else None,
267
+ assessment=MyApplicationsAssessment(
268
+ id=assessment.id if assessment else "",
269
+ title=assessment.title if assessment else "",
270
+ passing_score=assessment.passing_score if assessment else 0.0
271
+ ) if assessment else None,
272
+ score=score,
273
+ created_at=application.created_at.isoformat() if application.created_at else None
274
+ )
275
+
276
+ application_responses.append(application_response)
277
+
278
+ logger.info(f"Successfully retrieved {len(applications)} applications out of total {total} for user ID: {current_user.id}")
279
+ return MyApplicationsListResponse(
280
+ count=len(applications),
281
+ total=total,
282
+ data=application_responses
283
+ )
backend/models/application.py CHANGED
@@ -1,4 +1,5 @@
1
- from sqlalchemy import Column, String, Text, ForeignKey
 
2
  from .base import Base
3
  import uuid
4
 
@@ -9,4 +10,6 @@ class Application(Base):
9
  job_id = Column(String, ForeignKey("jobs.id"), nullable=False)
10
  assessment_id = Column(String, ForeignKey("assessments.id"), nullable=False)
11
  user_id = Column(String, ForeignKey("users.id"), nullable=False)
12
- answers = Column(Text) # Stored as JSON string
 
 
 
1
+ from sqlalchemy import Column, String, Text, ForeignKey, DateTime
2
+ from sqlalchemy.sql import func
3
  from .base import Base
4
  import uuid
5
 
 
10
  job_id = Column(String, ForeignKey("jobs.id"), nullable=False)
11
  assessment_id = Column(String, ForeignKey("assessments.id"), nullable=False)
12
  user_id = Column(String, ForeignKey("users.id"), nullable=False)
13
+ answers = Column(Text) # Stored as JSON string
14
+ created_at = Column(DateTime(timezone=True), server_default=func.now())
15
+ updated_at = Column(DateTime(timezone=True), onupdate=func.now())
backend/schemas/__init__.py CHANGED
@@ -1,11 +1,11 @@
1
  from .user import UserBase, UserCreate, UserUpdate, UserResponse, UserLogin, UserLogout, TokenResponse
2
  from .job import JobBase, JobCreate, JobUpdate, JobResponse, JobListResponse
3
  from .assessment import AssessmentBase, AssessmentCreate, AssessmentUpdate, AssessmentResponse, AssessmentListResponse, AssessmentDetailedResponse, AssessmentRegenerate
4
- from .application import ApplicationBase, ApplicationCreate, ApplicationUpdate, ApplicationResponse, ApplicationListResponse, ApplicationDetailedResponse, ApplicationDetailedListResponse
5
 
6
  __all__ = [
7
  "UserBase", "UserCreate", "UserUpdate", "UserResponse", "UserLogin", "UserLogout", "TokenResponse",
8
  "JobBase", "JobCreate", "JobUpdate", "JobResponse", "JobListResponse",
9
  "AssessmentBase", "AssessmentCreate", "AssessmentUpdate", "AssessmentResponse", "AssessmentListResponse", "AssessmentDetailedResponse", "AssessmentRegenerate",
10
- "ApplicationBase", "ApplicationCreate", "ApplicationUpdate", "ApplicationResponse", "ApplicationListResponse", "ApplicationDetailedResponse", "ApplicationDetailedListResponse"
11
  ]
 
1
  from .user import UserBase, UserCreate, UserUpdate, UserResponse, UserLogin, UserLogout, TokenResponse
2
  from .job import JobBase, JobCreate, JobUpdate, JobResponse, JobListResponse
3
  from .assessment import AssessmentBase, AssessmentCreate, AssessmentUpdate, AssessmentResponse, AssessmentListResponse, AssessmentDetailedResponse, AssessmentRegenerate
4
+ from .application import ApplicationBase, ApplicationCreate, ApplicationUpdate, ApplicationResponse, ApplicationListResponse, ApplicationDetailedResponse, ApplicationDetailedListResponse, MyApplicationsListResponse, MyApplicationResponse, MyApplicationsJob, MyApplicationsAssessment
5
 
6
  __all__ = [
7
  "UserBase", "UserCreate", "UserUpdate", "UserResponse", "UserLogin", "UserLogout", "TokenResponse",
8
  "JobBase", "JobCreate", "JobUpdate", "JobResponse", "JobListResponse",
9
  "AssessmentBase", "AssessmentCreate", "AssessmentUpdate", "AssessmentResponse", "AssessmentListResponse", "AssessmentDetailedResponse", "AssessmentRegenerate",
10
+ "ApplicationBase", "ApplicationCreate", "ApplicationUpdate", "ApplicationResponse", "ApplicationListResponse", "ApplicationDetailedResponse", "ApplicationDetailedListResponse", "MyApplicationsListResponse", "MyApplicationResponse", "MyApplicationsJob", "MyApplicationsAssessment"
11
  ]
backend/schemas/application.py CHANGED
@@ -64,4 +64,27 @@ class ApplicationListResponse(BaseModel):
64
  class ApplicationDetailedListResponse(BaseModel):
65
  count: int
66
  total: int
67
- data: List[ApplicationResponse]
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
64
  class ApplicationDetailedListResponse(BaseModel):
65
  count: int
66
  total: int
67
+ data: List[ApplicationResponse]
68
+
69
+ class MyApplicationsJob(BaseModel):
70
+ id: str
71
+ title: str
72
+ seniority: str
73
+ description: str
74
+
75
+ class MyApplicationsAssessment(BaseModel):
76
+ id: str
77
+ title: str
78
+ passing_score: float
79
+
80
+ class MyApplicationResponse(BaseModel):
81
+ id: str
82
+ job: MyApplicationsJob
83
+ assessment: MyApplicationsAssessment
84
+ score: float
85
+ created_at: Optional[str] = None
86
+
87
+ class MyApplicationsListResponse(BaseModel):
88
+ count: int
89
+ total: int
90
+ data: List[MyApplicationResponse]
backend/test_endpoint.py ADDED
@@ -0,0 +1,20 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import requests
2
+
3
+ # Test if the endpoint exists and returns proper error for unauthorized access
4
+ BASE_URL = "http://localhost:8000"
5
+
6
+ def test_unauthorized_access():
7
+ """Test that the endpoint requires authentication"""
8
+ response = requests.get(f"{BASE_URL}/applications/my-applications")
9
+
10
+ print(f"Status Code: {response.status_code}")
11
+ print(f"Response: {response.json()}")
12
+
13
+ # Should return 403 Forbidden since no token is provided
14
+ assert response.status_code == 403, f"Expected 403, got {response.status_code}"
15
+ print("OK - Endpoint correctly requires authentication")
16
+
17
+ if __name__ == "__main__":
18
+ print("Testing the new 'my-applications' endpoint...")
19
+ test_unauthorized_access()
20
+ print("Test completed successfully!")
backend/test_my_applications.py ADDED
@@ -0,0 +1,52 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import requests
2
+ import json
3
+
4
+ # Base URL for the API
5
+ BASE_URL = "http://localhost:8000"
6
+
7
+ def test_login():
8
+ """Test user login to get token"""
9
+ login_data = {
10
+ "email": "applicant@example.com", # Replace with actual test user
11
+ "password": "securepassword123" # Replace with actual test password
12
+ }
13
+
14
+ response = requests.post(f"{BASE_URL}/registration/login", json=login_data)
15
+
16
+ if response.status_code == 200:
17
+ token = response.json().get("access_token")
18
+ print(f"Login successful. Token: {token}")
19
+ return token
20
+ else:
21
+ print(f"Login failed: {response.status_code} - {response.text}")
22
+ return None
23
+
24
+ def test_get_my_applications(token):
25
+ """Test getting current user's applications"""
26
+ headers = {
27
+ "Authorization": f"Bearer {token}",
28
+ "Content-Type": "application/json"
29
+ }
30
+
31
+ response = requests.get(f"{BASE_URL}/applications/my-applications", headers=headers)
32
+
33
+ if response.status_code == 200:
34
+ applications = response.json()
35
+ print(f"My Applications retrieved successfully:")
36
+ print(json.dumps(applications, indent=2))
37
+ return applications
38
+ else:
39
+ print(f"Getting my applications failed: {response.status_code} - {response.text}")
40
+ return None
41
+
42
+ if __name__ == "__main__":
43
+ print("Testing the new 'my-applications' endpoint...")
44
+
45
+ # Login to get token
46
+ token = test_login()
47
+
48
+ if token:
49
+ # Test the new endpoint
50
+ test_get_my_applications(token)
51
+ else:
52
+ print("Skipping test due to login failure.")