Test Case: python-instruct-1643

  1import sqlite3  2import json  3import datetime  4import asyncio  5import websockets  6​  7async def update_database(data, websocket):  8    """  9    Updates a database using a websocket connection. 10​ 11    Args: 12        data (dict): A dictionary of data to update the database with. 13                     Expected keys: 14                         - table_name (str): The name of the table to update. 15                         - problem_id (int): The ID of the problem being updated. 16                         - solved_user (str): The username of the user who solved the problem. 17                         - other_fields (dict): A dictionary of other fields to update in the table. 18        websocket: The websocket connection object. 19    """ 20    try: 21        table_name = data['table_name'] 22        problem_id = data['problem_id'] 23        solved_user = data['solved_user'] 24        other_fields = data['other_fields'] 25​ 26        conn = sqlite3.connect('database.db')  # Replace with your database file 27        cursor = conn.cursor() 28​ 29        # Construct the SQL update query 30        update_fields = ', '.join([f"{key} = ?" for key in other_fields]) 31        sql = f"UPDATE {table_name} SET {update_fields} WHERE id = ?" 32        values = list(other_fields.values()) + [problem_id] 33​ 34        cursor.execute(sql, values) 35​ 36        # Increment the solved counter 37        cursor.execute(f"UPDATE {table_name} SET solved_count = solved_count + 1 WHERE id = ?", (problem_id,)) 38​ 39        # Update the solved user and last date 40        now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") 41        cursor.execute(f"UPDATE {table_name} SET solved_user = ?, last_solved = ? WHERE id = ?", (solved_user, now, problem_id)) 42​ 43        # Commit the changes 44        conn.commit() 45​ 46        # Send a success message back to the client 47        await websocket.send(json.dumps({"status": "success", "message": "Database updated successfully"})) 48​ 49    except Exception as e: 50        print(f"Error updating database: {e}") 51        conn.rollback() 52        await websocket.send(json.dumps({"status": "error", "message": str(e)})) 53​ 54    finally: 55        if conn: 56            conn.close() 57​ 58​ 59async def handle_connection(websocket, path): 60    """ 61    Handles a websocket connection and processes incoming messages. 62    """ 63    try: 64        async for message in websocket: 65            try: 66                data = json.loads(message) 67                await update_database(data, websocket) 68            except json.JSONDecodeError: 69                print("Received invalid JSON") 70                await websocket.send(json.dumps({"status": "error", "message": "Invalid JSON"})) 71            except Exception as e: 72                print(f"Error processing message: {e}") 73                await websocket.send(json.dumps({"status": "error", "message": str(e)})) 74    except websockets.exceptions.ConnectionClosedError as e: 75        print(f"Connection closed unexpectedly: {e}") 76    except Exception as e: 77        print(f"Error in handle_connection: {e}") 78​ 79​ 80async def main(): 81    """ 82    Starts the websocket server. 83    """ 84    async with websockets.serve(handle_connection, "localhost", 8765): 85        print("WebSocket server started at ws://localhost:8765") 86        await asyncio.Future()  # Run forever 87​ 88if __name__ == "__main__": 89    asyncio.run(main())

None

  1import json  2from sqlalchemy import create_engine, Column, Integer, String, DateTime, func  3from sqlalchemy.ext.declarative import declarative_base  4from sqlalchemy.orm import sessionmaker  5from sqlalchemy.orm import Session  6from datetime import datetime  7​  8Base = declarative_base()  9​ 10class Problem(Base): 11    __tablename__ = 'problems' 12​ 13    id = Column(Integer, primary_key=True) 14    name = Column(String) 15    description = Column(String) 16    solved_count = Column(Integer, default=0) 17    solved_user = Column(String, nullable=True) 18    last_solved_date = Column(DateTime, nullable=True) 19​ 20    def __repr__(self): 21        return f"<Problem(name='{self.name}', solved_count={self.solved_count})>" 22​ 23​ 24def update_database(db_url: str, data: dict) -> None: 25    """ 26    Updates a database using SQLAlchemy based on data received via a websocket connection. 27​ 28    Args: 29        db_url: The database connection string. 30        data: A dictionary containing the data to update.  It is expected to have 31              'problem_id' and 'solving_user' keys.  Other keys are assumed to 32              correspond to columns in the 'problems' table. 33    """ 34​ 35    engine = create_engine(db_url, hide_parameters=True)  # hide_parameters for security 36    Base.metadata.create_all(engine) 37    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) 38​ 39    with SessionLocal() as session: 40        try: 41            problem_id = data.get('problem_id') 42            solving_user = data.get('solving_user') 43​ 44            if not problem_id or not solving_user: 45                print("Error: 'problem_id' and 'solving_user' are required in the data.") 46                return  # Or raise an exception, depending on desired behavior 47​ 48            problem = session.query(Problem).filter(Problem.id == problem_id).first() 49​ 50            if not problem: 51                print(f"Error: Problem with id {problem_id} not found.") 52                return  # Or raise an exception 53​ 54            # Update other fields from the data dictionary (excluding problem_id and solving_user) 55            for key, value in data.items(): 56                if key not in ('problem_id', 'solving_user'): 57                    if hasattr(problem, key): 58                        setattr(problem, key, value) 59                    else: 60                        print(f"Warning: Column '{key}' not found in the Problem model.") 61​ 62            # Increment solved counter, update solved user, and last solved date 63            problem.solved_count += 1 64            problem.solved_user = solving_user 65            problem.last_solved_date = datetime.utcnow() 66​ 67            session.add(problem) 68            session.commit()  # Explicit commit within the context manager 69            print(f"Problem {problem_id} updated successfully.") 70​ 71        except Exception as e: 72            session.rollback() 73            print(f"Error updating database: {e}") 74            raise  # Re-raise the exception for handling upstream 75​ 76        finally: 77            # Session is automatically closed by the context manager 78            pass 79​ 80​ 81if __name__ == '__main__': 82    # Example usage (replace with your actual database URL) 83    db_url = "sqlite:///./test.db" 84​ 85    # Example data received from a websocket 86    example_data = { 87        'problem_id': 1, 88        'solving_user': 'test_user', 89        'description': 'Updated description', 90        'name': 'Updated Name' 91    } 92​ 93    # Create a test database and insert a problem 94    engine = create_engine(db_url, hide_parameters=True) 95    Base.metadata.create_all(engine) 96    SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) 97​ 98    with SessionLocal() as session: 99        problem = Problem(id=1, name='Initial Problem', description='Initial Description')100        session.add(problem)101        session.commit()102​103    update_database(db_url, example_data)104​105    # Verify the update106    with SessionLocal() as session:107        updated_problem = session.query(Problem).filter(Problem.id == 1).first()108        print(updated_problem)109        print(f"Solved Count: {updated_problem.solved_count}")110        print(f"Solved User: {updated_problem.solved_user}")111        print(f"Last Solved Date: {updated_problem.last_solved_date}")

None