小编Ale*_*and的帖子

我有以下DAG和两个SSHExecuteOperator任务.第一个任务执行返回参数的存储过程.第二个任务需要此参数作为输入.

请问如何从task1中推送的XCom中提取值,以便在task2中使用它？

from airflow import DAG
from datetime import datetime, timedelta
from airflow.contrib.hooks.ssh_hook import SSHHook
from airflow.contrib.operators.ssh_execute_operator import SSHExecuteOperator
from airflow.models import Variable

default_args = {
  'owner': 'airflow',
  'depends_on_past': False,
  'start_date': datetime.now(),
  'email': ['my@email.com'],
  'email_on_failure': True,
  'retries': 0
}

#server must be changed to point to the correct environment, to do so update DataQualitySSHHook variable in Airflow admin
DataQualitySSHHook = Variable.get('DataQualitySSHHook')
print('Connecting to: ' + DataQualitySSHHook)
sshHookEtl = SSHHook(conn_id=DataQualitySSHHook)
sshHookEtl.no_host_key_check = True 

#create dag
dag = DAG(
  'ed_data_quality_test-v0.0.3', #update version …

我正在使用 Python 包 Flask、SQLAlchemy、Graphene 和 Graphene-SQLAlchemy 构建 GraphQL API。我遵循了SQLAlchemy + Flask 教程。我能够执行查询和更改来创建记录。现在我想知道更新现有记录的最佳方法是什么。

这是我当前的脚本schema.py：

from graphene_sqlalchemy import SQLAlchemyObjectType
from database.batch import BatchOwner as BatchOwnerModel
import api_utils  # Custom methods to create records in database
import graphene


class BatchOwner(SQLAlchemyObjectType):
    """Batch owners."""
    class Meta:
        model = BatchOwnerModel
        interfaces = (graphene.relay.Node,)


class CreateBatchOwner(graphene.Mutation):
    """Create batch owner."""
    class Arguments:
        name = graphene.String()

    # Class attributes
    ok = graphene.Boolean()
    batch_owner = graphene.Field(lambda: BatchOwner)

    def mutate(self, info, name):
        record = {'name': name}
        api_utils.create('BatchOwner', …

我有一组Python脚本，它们以嵌套方式调用函数。对于这些功能中的每一个，我都有一个尝试，除了语句以捕获每个异常并打印它们。我想发送一封电子邮件警报，其中包含执行期间遇到的异常的完整序列。例：

import sys

def SendAlert(ErrorMessage):
    try:
        #send email alert with error message
        #[...]
    except:
        print(str(sys.exc_info()))
        return(sys.exc_info())

def ParentFunction():
    try:
        #call ChildFunction
        ChildResult = ChildFunction()

        #do stuff with ChildResult
        #[...]
        return ParentResult
    except:
        ErrorMessage = str(sys.exc_info())
        print(ErrorMessage)
        SendAlert(ErrorMessage)

def ChildFunction():
    try:
        #do stuff
        #[...]
        return ChildResult
    except:
        print(str(sys.exc_info()))
        return(sys.exc_info())

#main
if __name__ == '__main__':
    Result = ParentFunction()

上面的代码在发生错误的情况下将表现如下，ChildFunction其中错误是嵌套最多的函数：

• ChildFunction 遇到异常，它将打印它并将错误消息返回给 ParentFunction
• ParentFunction将失败，因为ChildResult包含错误消息而不是有效值
• ParentFunction 将触发和异常并在电子邮件警报中发送自己的错误消息

除了来自的错误消息外ParentFunction，我还希望电子邮件警报中包含来自的错误消息ChildFunction。请注意，我想避免在except语句中将ChildResult变量传递给SendAlert函数，ParentFunction因为在现实生活中，我的程序有很多嵌套函数，这意味着将每个函数的结果变量传递给except语句。

您将如何实现？有没有办法访问整个程序触发的完整错误序列？ …

我目前为我的项目定义了以下突变：

我的类PlanetAttribute 用于定义用作突变输入的石墨烯字段

class PlanetAttribute:
    name = graphene.String(required=True, description="Name of the planet.")
    rotation_period = graphene.String(default_value="unknown", description="Rotation period of the planet.")
    orbital_period = graphene.String(default_value="unknown", description="Orbital period of the planet.")
    diameter = graphene.String(default_value="unknown", description="Diameter of the planet.")
    climate = graphene.String(default_value="unknown", description="Climate period of the planet.")
    gravity = graphene.String(default_value="unknown", description="Gravity of the planet.")
    terrain = graphene.String(default_value="unknown", description="Terrain of the planet.")
    surface_water = graphene.String(default_value="unknown", description="Surface water of the planet.")
    population = graphene.String(default_value="unknown", description="Population of the planet.")
    url = graphene.String(default_value="unknown", description="URL of the planet …

我正在尝试使用JayDeBeApi包在Python 3 中连接到Teradata数据库。这是使用Flask和Flask-Restplus的 API 项目的一部分

# Install JVM
sudo apt-get install default-jre

# Create Python virtual environment
sudo apt-get install python3-venv
python3 -m venv jdbc
source jdbc/bin/activate

# Install Python packages in virtual environment
pip3 install --upgrade pip
pip3 install jaydebeapi
pip3 install flask

from flask import Flask
import jaydebeapi

app = Flask(__name__)


def get_jdbc_connection():
    connection = jaydebeapi.connect(
        'com.teradata.jdbc.TeraDriver',
        'jdbc:teradata://edw-dev.company.org',
        {'user': 'LOGIN', 'password': 'PASSWORD', 'tmode': 'TERA', …