cosmosdb 的计时器触发器无法正常工作

Question

我对我的函数应用程序“TimerTrigger”有疑问。

我开发了此功能来与 Telegram 机器人通信，以便在 API 请求后发送消息。

我在本地尝试过该功能应用程序，效果很好。但是，当我尝试使用 CosmosDB 存储信息时，遇到问题并且无法保存信息。

我已经设置了将我的应用程序与 telegram 和 Cosmosdb 连接所需的所有变量和内容

    try:
        database_obj  = client.get_database_client(database_name)
        await database_obj.read()
        return database_obj
    except exceptions.CosmosResourceNotFoundError:
        print("Creating database")
        return await client.create_database(database_name)
# </create_database_if_not_exists>
    
# Create a container
# Using a good partition key improves the performance of database operations.
# <create_container_if_not_exists>
async def get_or_create_container(database_obj, container_name):
    try:        
        todo_items_container = database_obj.get_container_client(container_name)
        await todo_items_container.read()   
        return todo_items_container
    except exceptions.CosmosResourceNotFoundError:
        print("Creating container with lastName as partition key")
        return await database_obj.create_container(
            id=container_name,
            partition_key=PartitionKey(path="/lastName"),
            offer_throughput=400)
    except exceptions.CosmosHttpResponseError:
        raise
# </create_container_if_not_exists>

async def populate_container_items(container_obj, items_to_create):
    # Add items to the container
    family_items_to_create = items_to_create
    # <create_item>
    for family_item in family_items_to_create:
        inserted_item = await container_obj.create_item(body=family_item)
        print("Inserted item for %s family. Item Id: %s" %(inserted_item['lastName'], inserted_item['id']))
    # </create_item>
# </method_populate_container_items>

async def read_items(container_obj, items_to_read):
    # Read items (key value lookups by partition key and id, aka point reads)
    # <read_item>
    for family in items_to_read:
        item_response = await container_obj.read_item(item=family['id'], partition_key=family['lastName'])
        request_charge = container_obj.client_connection.last_response_headers['x-ms-request-charge']
        print('Read item with id {0}. Operation consumed {1} request units'.format(item_response['id'], (request_charge)))
    # </read_item>
# </method_read_items>

# <method_query_items>
async def query_items(container_obj, query_text):
    # enable_cross_partition_query should be set to True as the container is partitioned
    # In this case, we do have to await the asynchronous iterator object since logic
    # within the query_items() method makes network calls to verify the partition key
    # definition in the container
    # <query_items>
    query_items_response = container_obj.query_items(
        query=query_text,
        enable_cross_partition_query=True
    )
    request_charge = container_obj.client_connection.last_response_headers['x-ms-request-charge']
    items = [item async for item in query_items_response]
    print('Query returned {0} items. Operation consumed {1} request units'.format(len(items), request_charge))
    # </query_items>
# </method_query_items>

async def run_sample():
    print('aaaa')
    print('sss {0}'.format(CosmosClient(endpoint,credential=key)))
    async with CosmosClient(endpoint, credential = key) as client:
        print('connected to db')
        try:
            database_obj = await get_or_create_db(client, database_name)
            # create a container
            container_obj = await get_or_create_container(database_obj, container_name)
            family_items_to_create = ["link", "ss", "s", "s"]
            await populate_container_items(container_obj, family_items_to_create)
            await read_items(container_obj, family_items_to_create)
            # Query these items using the SQL query syntax. 
            # Specifying the partition key value in the query allows Cosmos DB to retrieve data only from the relevant partitions, which improves performance
            query = "SELECT * FROM c "
            await query_items(container_obj, query)   
        except exceptions.CosmosHttpResponseError as e:
            print('\nrun_sample has caught an error. {0}'.format(e.message))
        finally:
            print("\nQuickstart complete")

async def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()
    
    asyncio.create_task(run_sample())
    logging.info(' sono partito')
    sendNews()
    if mytimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function ran at %s', utc_timestamp)

我已开始我的功能

func host start --port 7072

但我认为与数据库的连接出了问题，因为 console.log('connected to db') 没有被打印。

似乎所有与cosmosdb相关的操作都没有执行，如果有错误不知道如何解决。

我的终端中没有任何错误，但正如我所说，cosmosdb 似乎不起作用。

我不确定是否向您提供了所有必要的信息。感谢您的帮助。

Answer 1

我在使用异步函数时也遇到了同样的问题。当我使用非异步函数时，它对我有用。

作为参考，请检查此文档

我的代码：

TimeTrigger1/__init__.py

：

import datetime
import logging
import asyncio
import azure.functions as func
from azure.cosmos import cosmos_client
import azure.cosmos.exceptions as exceptions
from azure.cosmos.partition_key import PartitionKey

endpoint = "https://timercosmosdb.documents.azure.com/"
key = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
database_name = "ToDoList"
container_name = "Test"

def get_or_create_db(client,database_name):
    try:
        database_obj  = client.get_database_client(database_name)
        database_obj.read()
        return database_obj
    except exceptions.CosmosResourceNotFoundError:
        logging.info("Creating database")
        return client.create_database_if_not_exists(database_name)
    

def get_or_create_container(database_obj, container_name):
    try:        
        todo_items_container = database_obj.get_container_client(container_name)
        todo_items_container.read()   
        return todo_items_container
    except exceptions.CosmosResourceNotFoundError:
        logging.info("Creating container with lastName as partition key")
        return database_obj.create_container_if_not_exists(
            id=container_name,
            partition_key=PartitionKey(path="/id"),
            offer_throughput=400)
    except exceptions.CosmosHttpResponseError:
        raise


def populate_container_items(container_obj,items):
    inserted_item = container_obj.create_item(body=items)
    logging.info("Inserted item for %s family. Item Id: %s" %(inserted_item['lastName'], inserted_item['id']))

def read_items(container_obj,id):
        item_response = container_obj.read_item(item=id, partition_key=id)
        request_charge = container_obj.client_connection.last_response_headers['x-ms-request-charge']
        logging.info('Read item with id {0}. Operation consumed {1} request units'.format(item_response['id'], (request_charge)))

def query_items(container_obj, query_text):
    query_items_response = container_obj.query_items(
        query=query_text,
        enable_cross_partition_query=True
    )
    request_charge = container_obj.client_connection.last_response_headers['x-ms-request-charge']
    items = [item for item in query_items_response]
    logging.info('Query returned {0} items. Operation consumed {1} request units'.format(len(items), request_charge))

def run_sample():
    logging.info('aaaa')
    client = cosmos_client.CosmosClient(endpoint, key)
    logging.info('connected to db')
    try:
        id= "test"
        database_obj = get_or_create_db(client,database_name)

        container_obj = get_or_create_container(database_obj,container_name)
        item_dict = {
                "id": id,
                "lastName": "Shandilya",
                "firstName": "Vivek",
                "gender": "male",
                "age": 35
            }
        populate_container_items(container_obj,item_dict)
        read_items(container_obj,id)

        query = "SELECT * FROM c "
        query_items(container_obj, query)   
    except exceptions.CosmosHttpResponseError as e:
        logging.info('\nrun_sample has caught an error. {0}'.format(e.message))
    finally:
        logging.info("\nQuickstart complete")

def main(mytimer: func.TimerRequest) -> None:
    utc_timestamp = datetime.datetime.utcnow().replace(
        tzinfo=datetime.timezone.utc).isoformat()
    
    run_sample()
    logging.info(' sono partito')
    logging.info('Python timer trigger function ran at %s', utc_timestamp)

OUTPUT

:

Functions:

        TimerTrigger1: timerTrigger

For detailed output, run func with --verbose flag.
[2024-01-30T09:00:24.818Z] Executing 'Functions.TimerTrigger1' (Reason='Timer fired at 2024-01-30T14:30:24.7842979+05:30', Id=5499e180-4964-4d7e-b9f2-b024860945dd)
[2024-01-30T09:00:24.822Z] Trigger Details: UnscheduledInvocationReason: IsPastDue, OriginalSchedule: 2024-01-30T14:30:00.0000000+05:30
[2024-01-30T09:00:25.022Z] aaaa
[2024-01-30T09:00:26.387Z] connected to db
[2024-01-30T09:00:28.212Z] Inserted item for Shandilya family. Item Id: test
[2024-01-30T09:00:28.373Z] Read item with id test. Operation consumed 1 request units
[2024-01-30T09:00:28.546Z]
Quickstart complete
[2024-01-30T09:00:28.548Z] Python timer trigger function ran at 2024-01-30T09:00:25.008468+00:00
[2024-01-30T09:00:28.547Z]  sono partito
[2024-01-30T09:00:28.546Z] Query returned 1 items. Operation consumed 1 request units
[2024-01-30T09:00:28.592Z] Executed 'Functions.TimerTrigger1' (Succeeded, Id=5499e180-4964-4d7e-b9f2-b024860945dd, Duration=3793ms)
[2024-01-30T09:00:29.296Z] Host lock lease acquired by instance ID '000000000000000000000000AAE5F384'.

{
    "id": "test",
    "lastName": "Shandilya",
    "firstName": "Vivek",
    "gender": "male",
    "age": 35,
    "_rid": "ey58AO9yWqwCAAAAAAAAAA==",
    "_self": "dbs/ey58AA==/colls/ey58AO9yWqw=/docs/ey58AO9yWqwCAAAAAAAAAA==/",
    "_etag": "\"01001327-0000-1a00-0000-65b8baac0000\"",
    "_attachments": "attachments/",
    "_ts": 1706605228
}