如何可以访问安装软件包并在本地测试我的功能应用程序？

Question

I在Azure中创建了一个HTTP函数应用程序，并且当我不依赖其他软件包（例如psycopg2，azure.storage.storage.blob或panda）时，它运行良好。现在我继续得到：

ModuleNotFoundError: No module named 'psycopg2'
[2025-03-03T12:56:56.243Z] No job functions found. Try making your job classes and methods public. If you're using binding extensions (e.g. Azure Storage, ServiceBus, Timers, etc.) make sure you've called the registration method for the extension(s) in your startup code (e.g. builder.AddAzureStorage(), builder.AddServiceBus(), builder.AddTimers(), etc.).

当我发布它时，它可以使用，但在本地行动不佳。我需要在对我部署的邮政测试之前对当地的Postgres DB进行本地测试。 我使用了虚拟环境，并使用

pip install -r requirements.txt

安装了我的软件包，但我仍然遇到相同的错误。 这是我的要求.txt Azure功能 Azure-Storage-blob 熊猫 psycopg2二进制 OpenPyxl

I创建了一个示例HTTP触发函数，以在本地测试Postgres DB，并成功测试了它。

Answer 1

.venv\Scripts\activate

通过运行

pip install -r requirements.txt

来重新安装依赖项

检查是否通过运行

psycopg2

。

Quirlements.txt：

pip list

function_app.py：

azure-functions
azure-storage-blob
pandas
psycopg2-binary
openpyxl

-local.settings.json：

import logging
import psycopg2
import pandas as pd
from azure.storage.blob import BlobServiceClient
import azure.functions as func
import os
import json

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

@app.route(route="upload-data", methods=["POST"])
def main(req: func.HttpRequest) -> func.HttpResponse:
    try:
        req_body = req.get_json()
        data = req_body.get("data")
        if not data:
            return func.HttpResponse("Invalid input: 'data' field is required.", status_code=400)
        
        conn = psycopg2.connect(os.getenv("PostgresConnectionString"))
        cur = conn.cursor()
        cur.execute("""
            CREATE TABLE IF NOT EXISTS people (
                id SERIAL PRIMARY KEY,
                name VARCHAR(100),
                age INT
            )
        """)

        for entry in data:
            name, age = entry.get("name"), entry.get("age")
            if name and age is not None:
                cur.execute("SELECT COUNT(*) FROM people WHERE name=%s AND age=%s", (name, age))
                if cur.fetchone()[0] == 0:
                    cur.execute("INSERT INTO people (name, age) VALUES (%s, %s)", (name, age))
        conn.commit()

        cur.execute("SELECT name, age FROM people")
        rows = cur.fetchall()
        df = pd.DataFrame(rows, columns=["Name", "Age"]).drop_duplicates()
        logging.info(f"Data fetched from Postgres:\n{df}")

        blob_service = BlobServiceClient.from_connection_string(os.getenv("AzureWebJobsStorage"))
        container_name = "kamcontainer"
        blob_name = "people_data.csv"

        blob_client = blob_service.get_blob_client(container=container_name, blob=blob_name)
        csv_data = df.to_csv(index=False)
        blob_client.upload_blob(csv_data, overwrite=True)
        cur.close()
        conn.close()
        logging.info("Data successfully inserted into Postgres and uploaded to Blob Storage")
        return func.HttpResponse("Success! Data inserted into Postgres and uploaded to Blob Storage.", status_code=200)

    except Exception as e:
        logging.error(f"Error: {e}")
        return func.HttpResponse(f"Internal Server Error: {e}", status_code=500)

I发送了邮政请求，将数据插入到Postgres并将该数据保存到Blob存储中。

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "<StorageConneString>",
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "PostgresConnectionString": "postgresql://<UserName>:<Password>@<postgressName>.postgres.database.azure.com:5432/<DBName>"
  }
}

http://localhost:7071/api/upload-data

输出：

您可以使用

{
  "data": [
    { "name": "Kamali", "age": 25 },
    { "name": "Vyshu", "age": 30 }
  ]
}

命令或按下func start

.

来运行该函数。

振动斑点存储：