Function process_documents

app/routers/processing.py:112–156

(
    document_limit: int = Query(default=None, description="Limit on number of documents to process"),
    generateQuestions: bool = Query(default=False, description="Flag to generate questions"),
    generateSummaries: bool = Query(default=False, description="Flag to generate summaries"),
    current_user: User = Depends(get_current_user))

             , description="Process documents in the database"
             , summary="Process documents in the database that have not had pages, questions or summaries created. Helps illustrate async processing")
async def process_documents(
    document_limit: int = Query(default=None, description="Limit on number of documents to process"),
    generateQuestions: bool = Query(default=False, description="Flag to generate questions"),
    generateSummaries: bool = Query(default=False, description="Flag to generate summaries"),
    current_user: User = Depends(get_current_user)):
    logging.basicConfig(level=logging.INFO)
    
    # Setup neo4j driver
    driver = GraphDatabase.driver(AppConfig.NEO4J_URI, auth=(AppConfig.NEO4J_USER, AppConfig.NEO4J_PASSWORD))

    query = "MATCH (a:Document) WHERE NOT (a)-[:HAS_PAGE]->(:Page) and a.text <> '' and a.process=True RETURN a.uuid as uuid"
    if document_limit is not None:
        query += f" LIMIT {document_limit}"
    
    logging.info("Querying for documents to process.")
    document_ids = []
    with driver.session() as session:
        result = session.run(query)
        document_ids = [record['uuid'] for record in result]
    session.close()

    logging.info(f"Found {len(document_ids)} documents to process.")
    
    task_ids = []
    for document_id in document_ids:
        try:
            logging.info(f"Queueing document {document_id} for processing.")
            with driver.session() as session:
                result = session.run("MATCH (a:Document {uuid: $uuid}) RETURN a", {"uuid": document_id})
                document_data = result.single().value()
                text = document_data['text']
                # Pass the generateQuestions and generateSummaries flags to the task
                task = process_text_task.delay(text, document_id, generateQuestions, generateSummaries)
                task_ids.append(task.id)
                logging.info(f"Queued document {document_id} with task ID {task.id}")
        except Exception as e:
            logging.error(f"Failed to queue document {document_id}: {e}")
        finally:
            session.close()
            driver.close()

    return {
        "message": f"Processing started for {len(document_ids)} documents",
        "task_ids": task_ids
    }

## Returns an access token based on username and password
@router.post("/token", response_model=Token, description="Returns an access token", summary="Returns an access token", tags=["Users"])