因此,我的目标是在验证表单时检查dynamodb表中的值是否存在。
def pipeline_exists(self, pipeline_name):
ses = boto3.session.Session()
try:
dynamodb = ses.resource('dynamodb')
except Exception as exception:
message = 'Error on DynamoDB client'
logging.error("Dynamodb client session failed: %d %s", exception)
msg = " Dynamodb client session failed: " + str(exception)
raise exception
table_name = 'my_tbl'
table = dynamodb.Table(table_name)
response = table.query(
KeyConditionExpression=Key('pipeline_name').eq(pipeline_name)
)
print(response)
if response['count'] > 0:
raise validators.ValidationError('Pipeline ' + pipeline_name.data + ' already exists')
else:
True
class AddPipeline(Form):
pipeline_name = StringField('Pipeline Name', [validators.Length(min=10, max=250), pipeline_exists])
我得到的错误是:
TypeError: Unsupported type "<class 'wtforms.fields.core.UnboundField'>" for value "<UnboundField(StringField, (), {})>"