(collection_key, new_serp_urls)
| 48 | close_connection(connection) |
| 49 | |
| 50 | def update_record(collection_key, new_serp_urls): |
| 51 | connection, cursor = create_connection() |
| 52 | |
| 53 | if connection and cursor: |
| 54 | try: |
| 55 | # SQL query to update the serp_urls for a specific record |
| 56 | update_query = """ |
| 57 | UPDATE embeddings |
| 58 | SET serp_urls = %s, updated_at = CURRENT_TIMESTAMP |
| 59 | WHERE collection_key = %s; |
| 60 | """ |
| 61 | cursor.execute(update_query, (new_serp_urls, collection_key)) |
| 62 | connection.commit() |
| 63 | print("Record updated successfully.") |
| 64 | except psycopg2.Error as error: |
| 65 | print("Error with database operation:", error) |
| 66 | finally: |
| 67 | close_connection(connection) |
nothing calls this directly
no test coverage detected