feat: add course code verification on registration

Author: lpm0073

README.md

@@ -11,8 +11,10 @@
 [![License: AGPL v3](https://img.shields.io/badge/License-AGPL_v3-blue.svg)](https://www.gnu.org/licenses/agpl-3.0)
 [![hack.d Lawrence McDaniel](https://img.shields.io/badge/hack.d-Lawrence%20McDaniel-orange.svg)](https://lawrencemcdaniel.com)
 
-A Python command-line application that demonstrates how to use OpenAI Api function calling to drive an automated agentic workflow. Also demonstrates how
-to use Docker Compose to containerize your project.
+A Python command-line application that demonstrates how to use OpenAI Api function calling to drive an automated agentic workflow. Additionally, this repo demonstrates
+
+- how to use Docker Compose to containerize your project
+- how to leverage Pydantic for constructing complex JSON objects
 
 Python code is [located here](./app/)
 

app/function_schemas.py

@@ -11,7 +11,6 @@
 from openai.types.chat import (
     ChatCompletion,
     ChatCompletionAssistantMessageParam,
-    ChatCompletionFunctionToolParam,
     ChatCompletionMessage,
     ChatCompletionMessageFunctionToolCallParam,
     ChatCompletionSystemMessageParam,
@@ -90,7 +89,6 @@ def process_tool_calls(message: ChatCompletionMessage) -> list[str]:
         return functions_called
     for tool_call in message.tool_calls:
 
-        # For function calls, access via type checking
         if tool_call.type == "function":
             function_name = tool_call.function.name
             function_args = json.loads(tool_call.function.arguments)
@@ -101,7 +99,7 @@ def process_tool_calls(message: ChatCompletionMessage) -> list[str]:
                     type="function",
                     function={
                         "name": function_name,
-                        "arguments": tool_call.function.arguments,  # Keep as string, don't parse
+                        "arguments": tool_call.function.arguments,
                     },
                 )
             ]
@@ -113,10 +111,8 @@ def process_tool_calls(message: ChatCompletionMessage) -> list[str]:
             )
             logger.info("Function call detected: %s with args %s", function_name, function_args)
 
-            # Execute the function
             function_result = handle_function_call(function_name, function_args)
 
-            # Add the function result to the conversation
             tool_message = ChatCompletionToolMessageParam(
                 role="tool", content=function_result, tool_call_id=tool_call.id
             )
@@ -132,20 +128,13 @@ def process_tool_calls(message: ChatCompletionMessage) -> list[str]:
 def completion(prompt: str) -> tuple[ChatCompletion, list[str]]:
     """LLM text completion"""
 
-    # Set the OpenAI API key
-    # -------------------------------------------------------------------------
     openai.api_key = settings.OPENAI_API_KEY
-
-    # setup our text completion prompt
-    # -------------------------------------------------------------------------
     model = settings.OPENAI_API_MODEL
     temperature = settings.OPENAI_API_TEMPERATURE
     max_tokens = settings.OPENAI_API_MAX_TOKENS
     messages.append(ChatCompletionUserMessageParam(role="user", content=prompt))
     functions_called = []
 
-    # Call the OpenAI API
-    # -------------------------------------------------------------------------
     response = openai.chat.completions.create(
         model=model,
         messages=messages,
@@ -156,14 +145,10 @@ def completion(prompt: str) -> tuple[ChatCompletion, list[str]]:
     )
     logger.debug("Initial response: %s", response.model_dump())
 
-    # Check if the model wants to call a function
-    # -------------------------------------------------------------------------
     message = response.choices[0].message
-
     while message.tool_calls:
         functions_called = process_tool_calls(message)
 
-        # Make another API call to get the final response
         response = openai.chat.completions.create(
             model=model,
             messages=messages,

app/prompt.py

@@ -1,22 +1,53 @@
 # -*- coding: utf-8 -*-
 """Stackademy application with MySQL database integration."""
 
+from enum import Enum
 from typing import Any, Dict, List, Optional
 
 from openai.types.chat import ChatCompletionFunctionToolParam
+from pydantic import BaseModel, Field
 
 from app.database import db
 from app.exceptions import ConfigurationException
-from app.function_schemas import GetCoursesParams, RegisterCourseParams
 from app.logging_config import get_logger, setup_logging
 
 
+class StackademySpecializationArea(str, Enum):
+    """Available specialization areas for courses."""
+
+    AI = "AI"
+    MOBILE = "mobile"
+    WEB = "web"
+    DATABASE = "database"
+    NETWORK = "network"
+    NEURAL_NETWORKS = "neural networks"
+
+
+class StackademyGetCoursesParams(BaseModel):
+    """Parameters for the get_courses function."""
+
+    max_cost: Optional[float] = Field(
+        None, description="The maximum cost that a student is willing to pay for a course."
+    )
+    description: Optional[StackademySpecializationArea] = Field(
+        None, description="Areas of specialization for courses in the catalogue."
+    )
+
+
+class StackademyRegisterCourseParams(BaseModel):
+    """Parameters for the register_course function."""
+
+    course_code: str = Field(description="The unique code for the course.")
+    email: str = Field(description="The email address of the new user.")
+    full_name: str = Field(description="The full name of the new user.")
+
+
 # Initialize logging
 setup_logging()
 logger = get_logger(__name__)
 
 
-class StackademyApp:
+class Stackademy:
     """Main application class for Stackademy with database functionality."""
 
     def __init__(self):
@@ -30,7 +61,7 @@ def tool_factory_get_courses(self) -> ChatCompletionFunctionToolParam:
             function={
                 "name": "get_courses",
                 "description": "returns up to 10 rows of course detail data, filtered by the maximum cost a student is willing to pay for a course and the area of specialization.",
-                "parameters": GetCoursesParams.model_json_schema(),
+                "parameters": StackademyGetCoursesParams.model_json_schema(),
             },
         )
 
@@ -41,7 +72,7 @@ def tool_factory_register(self) -> ChatCompletionFunctionToolParam:
             function={
                 "name": "register_course",
                 "description": "Register a student in a course with the provided details.",
-                "parameters": RegisterCourseParams.model_json_schema(),
+                "parameters": StackademyRegisterCourseParams.model_json_schema(),
             },
         )
 
@@ -108,6 +139,24 @@ def get_courses(self, description: Optional[str] = None, max_cost: Optional[floa
             logger.error("Failed to retrieve courses: %s", e)
             return []
 
+    def verify_course(self, course_code: str) -> bool:
+        """
+        Verify if a course exists in the database.
+        Args:
+            course_code (str): The course code to verify
+        Returns:
+            bool: True if the course exists, False otherwise
+        """
+        query = "SELECT * FROM courses WHERE course_code = %s"
+        logger.info("verify_course() course_code: %s", course_code)
+        try:
+            result = self.db.execute_query(query, (course_code,))
+            return len(result) > 0
+        # pylint: disable=broad-except
+        except Exception as e:
+            logger.error("Failed to retrieve courses: %s", e)
+            return False
+
     def register_course(self, course_code: str, email: str, full_name: str) -> bool:
         """
         Register a user for a course.
@@ -121,6 +170,9 @@ def register_course(self, course_code: str, email: str, full_name: str) -> bool:
             bool: True if registration is successful, False otherwise
         """
         logger.info("Registering %s (%s) for course %s...", full_name, email, course_code)
+        if not self.verify_course(course_code):
+            logger.error("Course code %s does not exist.", course_code)
+            return False
         return True
 
 
@@ -131,7 +183,7 @@ def main():
 
     try:
         # Initialize the application
-        app = StackademyApp()
+        app = Stackademy()
 
         # Test database connection
         logger.info("Testing database connection...")
@@ -159,7 +211,7 @@ def main():
         logger.error("Application error: %s", e)
 
 
-stackademy_app = StackademyApp()
+stackademy_app = Stackademy()
 
 if __name__ == "__main__":
     main()

app/response_models.py

@@ -1,21 +1,46 @@
 # -*- coding: utf-8 -*-
-"""Example of using OpenAI's structured outputs with Pydantic models."""
+"""EXPERIMENTAL: OpenAI's structured outputs with Pydantic models."""
 
 import json
-from typing import Optional
+from typing import List, Optional
 
 import openai
-from pydantic import ValidationError
+from pydantic import BaseModel, Field, ValidationError
 
 from app import settings
-from app.function_schemas import (
-    GetCoursesParams,
-    RegisterCourseParams,
-    SpecializationArea,
-)
 from app.logging_config import get_logger
-from app.response_models import Course, CourseSearchResponse, RegistrationResponse
-from app.stackademy import stackademy_app
+from app.stackademy import (
+    StackademyGetCoursesParams,
+    StackademyRegisterCourseParams,
+    StackademySpecializationArea,
+    stackademy_app,
+)
+
+
+class Course(BaseModel):
+    """A course in the Stackademy catalog."""
+
+    course_code: str = Field(description="The unique code for the course")
+    course_name: str = Field(description="The name of the course")
+    description: str = Field(description="Course description")
+    cost: float = Field(description="Cost of the course")
+    prerequisite_course_code: Optional[str] = Field(description="Prerequisite course code", default=None)
+    prerequisite_course_name: Optional[str] = Field(description="Prerequisite course name", default=None)
+
+
+class CourseSearchResponse(BaseModel):
+    """Response model for course search results."""
+
+    courses: List[Course] = Field(description="List of courses matching the search criteria")
+    total_count: int = Field(description="Total number of courses found")
+
+
+class RegistrationResponse(BaseModel):
+    """Response model for course registration."""
+
+    success: bool = Field(description="Whether the registration was successful")
+    message: str = Field(description="Human-readable message about the registration result")
+    registration_id: Optional[str] = Field(description="Unique registration ID if successful", default=None)
 
 
 logger = get_logger(__name__)
@@ -35,13 +60,13 @@ def get_courses_with_structured_output(
         if description:
 
             try:
-                specialization_area = SpecializationArea(description)
+                specialization_area = StackademySpecializationArea(description)
             except ValueError:
                 logger.warning("Invalid specialization area: %s", description)
                 specialization_area = None
 
         # Validate input parameters using Pydantic
-        params = GetCoursesParams(description=specialization_area, max_cost=max_cost)
+        params = StackademyGetCoursesParams(description=specialization_area, max_cost=max_cost)
 
         # Get raw course data
         courses_data = stackademy_app.get_courses(
@@ -68,7 +93,7 @@ def register_course_with_structured_output(course_code: str, email: str, full_na
     """
     try:
         # Validate input parameters
-        params = RegisterCourseParams(course_code=course_code, email=email, full_name=full_name)
+        params = StackademyRegisterCourseParams(course_code=course_code, email=email, full_name=full_name)
 
         # Attempt registration
         success = stackademy_app.register_course(