[Google App Engine][Python]Amazon Product Advertising APIで謎の503のエラーが発生する

AmazonのProduct Advertising APIでは1秒間に1回、1時間に3600回とリクエスト制限が決まっているのは知っていて、1時間に3600回リクエストも守っているのに、しかも、Pythonでtime.sleep(1)を入れているにもかかわらず、なぜか503エラーが発生します。

エラーの詳細を確認するともっとスローダウンさせて問い合わせをしてほしいって書いてあるんだけど、こちらとしては十分に寝かしているつもりなのだが。

ぐぐってみるとどうやら他の人も同じような悩みを抱えているみたいです。

もっと、sleepの間隔を広くとれってことなのかなー？

[Google App Engine][Python]Amazon Product Advertising API browsenodelookupを追加

ちょこちょこ修正を加えているAmazon Product Advertising APIを検索するPythonコードですが、以下のような形でbrowsenodelookupを検索するコードを追加しました。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2007 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

from urlparse import urlparse
from google.appengine.api import urlfetch
from lxml import etree

import os
import urllib
import datetime
import hashlib
import base64
import logging
import hmac
import xmltodict

class AmazonProductAdvertisingAPI(object):
  def __init__(self,access_key_id,secret_access_key,associate_tag = None,locale = "US"):

    self._access_key_id = access_key_id
    self._secret_access_key = secret_access_key
    self._associate_tag = associate_tag

    self._validate_certificate = True
    _environ = os.environ
    _http_host = _environ["HTTP_HOST"]
    if _http_host.find("localhost") > -1:
      self._validate_certificate = False

    _wsgi_url_scheme = _environ["wsgi.url_scheme"]
    self._endpoint = {
      "JP":_wsgi_url_scheme + "://webservices.amazon.co.jp/onca/xml",
      "US":_wsgi_url_scheme + "://webservices.amazon.com/onca/xml",
      "FR":_wsgi_url_scheme + "://webservices.amazon.fr/onca/xml",
      "IN":_wsgi_url_scheme + "://webservices.amazon.in/onca/xml",
      "IT":_wsgi_url_scheme + "://webservices.amazon.it/onca/xml",
      "UK":_wsgi_url_scheme + "://webservices.amazon.co.uk/onca/xml"
    }

    if self._endpoint.has_key(locale):
      self._product_advertising_url = self._endpoint[locale]
    else:
      self._product_advertising_url = self._endpoint["US"]

    self._service = "AWSECommerceService"

    #self._method = ""
    #self._prms = {}

  #タイムスタンプを設定
  def _timestamp(self,prms):

    prms["Timestamp"] = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")

  #URLエンコード
  def _urlencode_rfc3986(self,s):
    _s = urllib.quote(s)
    _s = _s.replace("%7E", "~")
    return _s

  #アクセストークン用の文字列作成
  def _canonical_string(self,prms):

    _keys = sorted(prms)
    _canonical_string = ""
    for _k in _keys:
      _canonical_string = _canonical_string + "&" + self._urlencode_rfc3986(_k) + "=" + self._urlencode_rfc3986(prms[_k])
    else:
      _canonical_string = _canonical_string[1:]

    return _canonical_string

  #問い合わせURLの作成
  def _request_url(self,prms,method):

    _url_info = urlparse(self._product_advertising_url)

    _canonical_string = self._canonical_string(prms)

    _signature = method + "\n" + _url_info.hostname + "\n" + _url_info.path +"\n" + _canonical_string
    _signature = hmac.new(self._secret_access_key, _signature, hashlib.sha256).digest()
    _signature = base64.b64encode(_signature)

    _url = self._product_advertising_url + "?" + _canonical_string + "&Signature=" + self._urlencode_rfc3986(_signature)

    return _url

  #itemsearch
  def itemsearch(self,searchindex,browsenode=None,availability="Available",responsegroup="Small",itempage=1,keywords=None,minimumprice=None,maximumprice=None,isremoveadult=True,sort=None):

    _prms = {}

    _prms["Service"] = self._service
    _prms["AWSAccessKeyId"] = self._access_key_id

    if self._associate_tag is not None:
      _prms["AssociateTag"] = self._associate_tag

    _prms["Operation"] = "ItemSearch"
    self._timestamp(_prms)
    
    if availability is None:
      return None

    for i in ["Available"]:
      if i == availability:
        _prms["Availability"] = availability
        break
    else:
      return None

    if searchindex is None:
      return None

    if searchindex == u"":
      return None

    _prms["SearchIndex"] = searchindex

    if responsegroup is None:
      return None

    if responsegroup == "":
      return None

    _prms["ResponseGroup"] = responsegroup

    if browsenode is not None:
      _prms["BrowseNode"] = browsenode

    if itempage is not None:
      _prms["ItemPage"] = str(itempage)

    if keywords is not None:
      #logging.info()
      if isinstance(keywords,unicode):
        #logging.info(keywords)
        _prms["Keywords"] = keywords.encode('utf-8')
      else:
        _prms["Keywords"] = keywords
      #_prms["Keywords"] = keywords

    if maximumprice is not None:
      _prms["MaximumPrice"] = str(maximumprice)

    if minimumprice is not None:
      _prms["MinimumPrice"] = str(minimumprice)

    if sort is not None:
      _prms["Sort"] = sort


    _url = self._request_url(_prms,"GET")
    #logging.info(_url)
    _result = self._request(_url)
    return _result

  #itemlookup
  def itemlookup(self,asin,responsegroup="Small"):
    _prms = {}

    _prms["Service"] = self._service
    _prms["AWSAccessKeyId"] = self._access_key_id

    if self._associate_tag is not None:
      _prms["AssociateTag"] = self._associate_tag

    _prms["Operation"] = "ItemLookup"
    self._timestamp(_prms)

    if asin is None:
      return None

    if asin == "":
      return None

    _prms["ItemId"] = asin

    if responsegroup is None:
      return None

    if responsegroup == "":
      return None

    _prms["ResponseGroup"] = responsegroup

    _url = self._request_url(_prms,"GET")
    _result = self._request(_url)
    return _result

  def _request(self,url):
    try:
      _result = urlfetch.fetch(url,validate_certificate = self._validate_certificate)
      _status_code = _result.status_code
      #logging.info(_status_code)

      if _status_code == 200:

        _content = _result.content
        _root = etree.fromstring(_content)
        _ns = _root.xpath('namespace-uri(.)')
        _namespace = {"ns":_ns}
        _errorTags = _root.findall('.//ns:Error',namespaces=_namespace)
        #logging.info(_content)
        if len(_errorTags) > 0:
          _errorTag = _errorTags[0]
          _code  = _errorTag.findall('.//ns:Code',namespaces=_namespace)
          _message = _errorTag.findall('.//ns:Message',namespaces=_namespace)
          logging.error(u"code: " + _code[0].text + " message: " + _message[0].text)
        else:
          return _root
      else:
        logging.error(u"status code: " + str(_status_code) + " content:" + _content)
        
    except urlfetch.DeadlineExceededError:
      logging.error(u"urlfetch DeadlineExceededError")

    except urlfetch.ResponseTooLargeError, response:
      logging.error(u"urlfetch ResponseTooLargeError" + response)

    except urlfetch.InternalTransientError:
      logging.error(u"urlfetch InternalTransientError")

    except urlfetch.InvalidURLError:
      logging.error(u"urlfetch InvalidURLError")

    except urlfetch.Error:
      logging.error(u"urlfetch Error")

    return None


  def browsenodelookup(self,browsenodeid):

      _prms = {}

      _prms["Service"] = self._service
      _prms["AWSAccessKeyId"] = self._access_key_id

      if browsenodeid is None or browsenodeid == "":
        return None

      if isinstance(browsenodeid,unicode):
        _prms["BrowseNodeId"] = browsenodeid.encode('utf-8')
      else:
        _prms["BrowseNodeId"] = browsenodeid

      if self._associate_tag is not None:
        _prms["AssociateTag"] = self._associate_tag

      _prms["Operation"] = "BrowseNodeLookup"
      self._timestamp(_prms)

      _url = self._request_url(_prms,"GET")
      #logging.info(_url)
      _result = self._request(_url)
      return _result

[Google App Engine][Python]Amazon Product Advertising APIをリファクタリング

ちょこちょこ作り続けているGoogle App EngineのPythonでAmazon Product Advertising APIをコールするクラスですが、itemsearchとitemlookupで同じ処理が2回呼ばれている部分があり保守性が悪かったので下記のようにリファクタリングしました。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2007 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

from urlparse import urlparse
from google.appengine.api import urlfetch
from lxml import etree

import os
import urllib
import datetime
import hashlib
import base64
import logging
import hmac
import xmltodict

class AmazonProductAdvertisingAPI(object):
  def __init__(self,access_key_id,secret_access_key,associate_tag = None,locale = "US"):

    self._access_key_id = access_key_id
    self._secret_access_key = secret_access_key
    self._associate_tag = associate_tag

    self._validate_certificate = True
    _environ = os.environ
    _http_host = _environ["HTTP_HOST"]
    if _http_host.find("localhost") > -1:
      self._validate_certificate = False

    _wsgi_url_scheme = _environ["wsgi.url_scheme"]
    self._endpoint = {
      "JP":_wsgi_url_scheme + "://webservices.amazon.co.jp/onca/xml",
      "US":_wsgi_url_scheme + "://webservices.amazon.com/onca/xml",
      "FR":_wsgi_url_scheme + "://webservices.amazon.fr/onca/xml",
      "IN":_wsgi_url_scheme + "://webservices.amazon.in/onca/xml",
      "IT":_wsgi_url_scheme + "://webservices.amazon.it/onca/xml",
      "UK":_wsgi_url_scheme + "://webservices.amazon.co.uk/onca/xml"
    }

    if self._endpoint.has_key(locale):
      self._product_advertising_url = self._endpoint[locale]
    else:
      self._product_advertising_url = self._endpoint["US"]

    self._service = "AWSECommerceService"

    #self._method = ""
    #self._prms = {}

  #タイムスタンプを設定
  def _timestamp(self,prms):

    prms["Timestamp"] = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")

  #URLエンコード
  def _urlencode_rfc3986(self,s):
    _s = urllib.quote(s)
    _s = _s.replace("%7E", "~")
    return _s

  #アクセストークン用の文字列作成
  def _canonical_string(self,prms):

    _keys = sorted(prms)
    _canonical_string = ""
    for _k in _keys:
      _canonical_string = _canonical_string + "&" + self._urlencode_rfc3986(_k) + "=" + self._urlencode_rfc3986(prms[_k])
    else:
      _canonical_string = _canonical_string[1:]

    return _canonical_string

  #問い合わせURLの作成
  def _request_url(self,prms,method):

    _url_info = urlparse(self._product_advertising_url)

    _canonical_string = self._canonical_string(prms)

    _signature = method + "\n" + _url_info.hostname + "\n" + _url_info.path +"\n" + _canonical_string
    _signature = hmac.new(self._secret_access_key, _signature, hashlib.sha256).digest()
    _signature = base64.b64encode(_signature)

    _url = self._product_advertising_url + "?" + _canonical_string + "&Signature=" + self._urlencode_rfc3986(_signature)

    return _url

  #itemsearch
  def itemsearch(self,searchindex,browsenode=None,availability="Available",responsegroup="Small",itempage=1,keywords=None,minimumprice=None,maximumprice=None,isremoveadult=True,sort=None):

    _prms = {}

    _prms["Service"] = self._service
    _prms["AWSAccessKeyId"] = self._access_key_id

    if self._associate_tag is not None:
      _prms["AssociateTag"] = self._associate_tag

    _prms["Operation"] = "ItemSearch"
    self._timestamp(_prms)
    
    if availability is None:
      return None

    for i in ["Available"]:
      if i == availability:
        _prms["Availability"] = availability
        break
    else:
      return None

    if searchindex is None:
      return None

    if searchindex == u"":
      return None

    _prms["SearchIndex"] = searchindex

    if responsegroup is None:
      return None

    if responsegroup == "":
      return None

    _prms["ResponseGroup"] = responsegroup

    if browsenode is not None:
      _prms["BrowseNode"] = browsenode

    if itempage is not None:
      _prms["ItemPage"] = str(itempage)

    if keywords is not None:
      #logging.info()
      if isinstance(keywords,unicode):
        #logging.info(keywords)
        _prms["Keywords"] = keywords.encode('utf-8')
      else:
        _prms["Keywords"] = keywords
      #_prms["Keywords"] = keywords

    if maximumprice is not None:
      _prms["MaximumPrice"] = str(maximumprice)

    if minimumprice is not None:
      _prms["MinimumPrice"] = str(minimumprice)

    if sort is not None:
      _prms["Sort"] = sort


    _url = self._request_url(_prms,"GET")
    #logging.info(_url)
    _result = self._request(_url)
    return _result

  #itemlookup
  def itemlookup(self,asin,responsegroup="Small"):
    _prms = {}

    _prms["Service"] = self._service
    _prms["AWSAccessKeyId"] = self._access_key_id

    if self._associate_tag is not None:
      _prms["AssociateTag"] = self._associate_tag

    _prms["Operation"] = "ItemLookup"
    self._timestamp(_prms)

    if asin is None:
      return None

    if asin == "":
      return None

    _prms["ItemId"] = asin

    if responsegroup is None:
      return None

    if responsegroup == "":
      return None

    _prms["ResponseGroup"] = responsegroup

    _url = self._request_url(_prms,"GET")
    _result = self._request(_url)
    return _result

  def _request(self,url):
    try:
      _result = urlfetch.fetch(url,validate_certificate = self._validate_certificate)
      _status_code = _result.status_code
      #logging.info(_status_code)

      if _status_code == 200:

        _content = _result.content
        _root = etree.fromstring(_content)
        _ns = _root.xpath('namespace-uri(.)')
        _namespace = {"ns":_ns}
        _errorTags = _root.findall('.//ns:Error',namespaces=_namespace)
        #logging.info(_content)
        if len(_errorTags) > 0:
          _errorTag = _errorTags[0]
          _code  = _errorTag.findall('.//ns:Code',namespaces=_namespace)
          _message = _errorTag.findall('.//ns:Message',namespaces=_namespace)
          logging.error(u"code: " + _code[0].text + " message: " + _message[0].text)
        else:
          return _root
      else:
        logging.error(u"status code: " + str(_status_code) + " content:" + _content)
        
    except urlfetch.DeadlineExceededError:
      logging.error(u"urlfetch DeadlineExceededError")

    except urlfetch.ResponseTooLargeError, response:
      logging.error(u"urlfetch ResponseTooLargeError" + response)

    except urlfetch.InternalTransientError:
      logging.error(u"urlfetch InternalTransientError")

    except urlfetch.InvalidURLError:
      logging.error(u"urlfetch InvalidURLError")

    except urlfetch.Error:
      logging.error(u"urlfetch Error")

    return None

[Google App Engine][Python]管理者の確認

Google App Engine上で、管理者かどうか確認する場合はどうすればいいのだろうか？

調査してみると以下のコードでおｋみたい。

from google.appengine.api import users
if users.get_current_user():
  if users.is_current_user_admin():
    logging.info(u"管理者")
  else:
    pass
else:
  pass

[Google App Engine][Python]Amazon Product Advertising APIのitemlookupを追加

先日、Google App EngineのPythonを使ってAmazon Product Advertising APIをclass化してみたのですが、新規にitemlookupを追加しました。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2007 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

from urlparse import urlparse
from google.appengine.api import urlfetch
from lxml import etree

import os
import urllib
import datetime
import hashlib
import base64
import logging
import hmac
import xmltodict

class AmazonProductAdvertisingAPI(object):
  def __init__(self,access_key_id,secret_access_key,associate_tag = None,locale = "US"):

    self._access_key_id = access_key_id
    self._secret_access_key = secret_access_key
    self._associate_tag = associate_tag

    self._validate_certificate = True
    _environ = os.environ
    _http_host = _environ["HTTP_HOST"]
    if _http_host.find("localhost") > -1:
      self._validate_certificate = False

    _wsgi_url_scheme = _environ["wsgi.url_scheme"]
    self._endpoint = {
      "JP":_wsgi_url_scheme + "://webservices.amazon.co.jp/onca/xml",
      "US":_wsgi_url_scheme + "://webservices.amazon.com/onca/xml",
      "FR":_wsgi_url_scheme + "://webservices.amazon.fr/onca/xml",
      "IN":_wsgi_url_scheme + "://webservices.amazon.in/onca/xml",
      "IT":_wsgi_url_scheme + "://webservices.amazon.it/onca/xml",
      "UK":_wsgi_url_scheme + "://webservices.amazon.co.uk/onca/xml"
    }

    if self._endpoint.has_key(locale):
      self._product_advertising_url = self._endpoint[locale]
    else:
      self._product_advertising_url = self._endpoint["US"]

    self._service = "AWSECommerceService"

    #self._method = ""
    #self._prms = {}

  #タイムスタンプを設定
  def _timestamp(self,prms):

    prms["Timestamp"] = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")

  #URLエンコード
  def _urlencode_rfc3986(self,s):
    _s = urllib.quote(s)
    _s = _s.replace("%7E", "~")
    return _s

  #アクセストークン用の文字列作成
  def _canonical_string(self,prms):

    _keys = sorted(prms)
    _canonical_string = ""
    for _k in _keys:
      _canonical_string = _canonical_string + "&" + self._urlencode_rfc3986(_k) + "=" + self._urlencode_rfc3986(prms[_k])
    else:
      _canonical_string = _canonical_string[1:]

    return _canonical_string

  #問い合わせURLの作成
  def _request_url(self,prms,method):

    _url_info = urlparse(self._product_advertising_url)

    _canonical_string = self._canonical_string(prms)

    _signature = method + "\n" + _url_info.hostname + "\n" + _url_info.path +"\n" + _canonical_string
    _signature = hmac.new(self._secret_access_key, _signature, hashlib.sha256).digest()
    _signature = base64.b64encode(_signature)

    _url = self._product_advertising_url + "?" + _canonical_string + "&Signature=" + self._urlencode_rfc3986(_signature)

    return _url

  #itemsearch
  def itemsearch(self,searchindex=None,browsenode=None,availability="Available",responsegroup="Small",itempage=1,keywords=None,minimumprice=None,maximumprice=None,isremoveadult=True,sort=None):

    _prms = {}

    _prms["Service"] = self._service
    _prms["AWSAccessKeyId"] = self._access_key_id

    if self._associate_tag is not None:
      _prms["AssociateTag"] = self._associate_tag

    _prms["Operation"] = "ItemSearch"
    self._timestamp(_prms)
    
    if availability == "":
      return None

    if availability is None:
      return None

    _prms["Availability"] = availability

    if searchindex == "":
      return None

    if searchindex is None:
      return None

    _prms["SearchIndex"] = searchindex

    if responsegroup is None:
      return None

    if responsegroup == "":
      return None

    _prms["ResponseGroup"] = responsegroup

    if browsenode is not None:
      _prms["BrowseNode"] = browsenode

    if itempage is not None:
      _prms["ItemPage"] = str(itempage)

    if keywords is not None:
      _prms["Keywords"] = keywords

    if maximumprice is not None:
      _prms["MaximumPrice"] = str(maximumprice)

    if minimumprice is not None:
      _prms["MinimumPrice"] = str(minimumprice)

    if sort is not None:
      _prms["Sort"] = sort

    _url = self._request_url(_prms,"GET")

    try:
      _result = urlfetch.fetch(_url,validate_certificate = self._validate_certificate)
      _status_code = _result.status_code
      #logging.info(_status_code)


      if _status_code == 200:

        _content = _result.content
        _root = etree.fromstring(_content)
        _ns = _root.xpath('namespace-uri(.)')
        _namespace = {"ns":_ns}
        _errorTags = _root.findall('.//ns:Error',namespaces=_namespace)
        #logging.info(_content)
        if len(_errorTags) > 0:
          _errorTag = _errorTags[0]
          _code  = _errorTag.findall('.//ns:Code',namespaces=_namespace)
          _message = _errorTag.findall('.//ns:Message',namespaces=_namespace)
          logging.error(u"code: " + _code[0].text + " message: " + _message[0].text)
        else:
          return _root
      else:
        logging.error(u"status code: " + str(_status_code) + " content:" + _content)
        
    except urlfetch.DeadlineExceededError:
      logging.error(u"urlfetch DeadlineExceededError")

    except urlfetch.ResponseTooLargeError, response:
      logging.error(u"urlfetch ResponseTooLargeError" + response)

    except urlfetch.InternalTransientError:
      logging.error(u"urlfetch InternalTransientError")

    except urlfetch.InvalidURLError:
      logging.error(u"urlfetch InvalidURLError")

    except urlfetch.Error:
      logging.error(u"urlfetch Error")

    return None

  #itemlookup
  def itemlookup(self,asin,responsegroup="Small"):
    _prms = {}

    _prms["Service"] = self._service
    _prms["AWSAccessKeyId"] = self._access_key_id

    if self._associate_tag is not None:
      _prms["AssociateTag"] = self._associate_tag

    _prms["Operation"] = "ItemLookup"
    self._timestamp(_prms)

    if asin is None:
      return None

    if asin == "":
      return None

    _prms["ItemId"] = asin

    if responsegroup is None:
      return None

    if responsegroup == "":
      return None

    _prms["ResponseGroup"] = responsegroup

    _url = self._request_url(_prms,"GET")

    try:
      _result = urlfetch.fetch(_url,validate_certificate = self._validate_certificate)
      _status_code = _result.status_code
      #logging.info(_status_code)


      if _status_code == 200:

        _content = _result.content
        _root = etree.fromstring(_content)
        _ns = _root.xpath('namespace-uri(.)')
        _namespace = {"ns":_ns}
        _errorTags = _root.findall('.//ns:Error',namespaces=_namespace)
        #logging.info(_content)
        if len(_errorTags) > 0:
          _errorTag = _errorTags[0]
          _code  = _errorTag.findall('.//ns:Code',namespaces=_namespace)
          _message = _errorTag.findall('.//ns:Message',namespaces=_namespace)
          logging.error(u"code: " + _code[0].text + " message: " + _message[0].text)
        else:
          return _root
      else:
        logging.error(u"status code: " + str(_status_code) + " content:" + _content)
        
    except urlfetch.DeadlineExceededError:
      logging.error(u"urlfetch DeadlineExceededError")

    except urlfetch.ResponseTooLargeError, response:
      logging.error(u"urlfetch ResponseTooLargeError" + response)

    except urlfetch.InternalTransientError:
      logging.error(u"urlfetch InternalTransientError")

    except urlfetch.InvalidURLError:
      logging.error(u"urlfetch InvalidURLError")

    except urlfetch.Error:
      logging.error(u"urlfetch Error")

    return None

[Google App Engine][Python]ItemPageの上限

Product Advertising APIのItemPageの上限はどうやら10までみたい

泣きそう。

[Google App Engine][Python]xmltodict

lxml.etreeを使ってxmlを毎回、毎回、パースするのがめんどくさいので、xmlをjsonライクに変換してくれるライブラリーってないのかなーっと思って探していたら、どうやら
xmltodict
を発見して、以下のようにコードを組んだらjsonライクなdictonaryに変換されました。

import xmltodict
import json
from lxml import etree

_xml = etree.tostring("")
_json = json.dumps(xmltodict.parse(_xml))

[Python][Google App Engine]Google Cloud Storageにファイルを読み込む

昨日のエントリーではGoogle App EngineからGoogle Cloud Storageにファイルをアップロードする方法について書いたので、そのファイルをreadする方法を書きたいと思います。

import os
import cloudstorage
from google.appengine.api import app_identity

bucket_name = os.environ.get(DEFAULT_BUCKET_NAME,app_identity.get_default_gcs_bucket_name())
filename = '/' + bucket_name + '/' + "hoge"
gcs_file = cloudstorage.open(filename)
contents = gcs_file.read()
gcs_file.close()

確かにreadできたー。

[Python][Google App Engine]Google Cloud Storageにファイルをアップロード

どうやらGoogle App EngineからGoogle Cloud Storageに5GBまでファイル保存が可能なようなので、アップロードの仕方を調査しました。

import os
import cloudstorage
from google.appengine.api import app_identity

bucket_name = os.environ.get(DEFAULT_BUCKET_NAME,app_identity.get_default_gcs_bucket_name())
write_retry_params = cloudstorage.RetryParams(backoff_factor=1)
filename = '/' + bucket_name + '/' + "hoge"
gcs_file = cloudstorage.open(
  filename,
  mode='w',
  content_type='text/plain',
  options={'x-goog-acl':'private'},
  retry_params=write_retry_params
)
gcs_file.write("foo bar")
gcs_file.close()

で実行したら確かにファイルが保存された超うれしい！

[Google App Engine][Python]Amazon Product Advertising APIをclass化してみた

だいぶ前に、AmazonのProduct　Advertising　APIでitemsearchをする方法を書いたような気がしたのですが、その時に、classではなくプロシージャで書いてしまい非常に可読性が悪かったので、下のようにclass化してみた。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2007 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

from urlparse import urlparse
from google.appengine.api import urlfetch
from lxml import etree

import os
import urllib
import datetime
import hashlib
import base64
import logging
import hmac

class AmazonProductAdvertisingAPI(object):
  def __init__(self,access_key_id,secret_access_key,associate_tag = None,locale = "US"):

    self._access_key_id = access_key_id
    self._secret_access_key = secret_access_key
    self._associate_tag = associate_tag

    self._validate_certificate = True
    _environ = os.environ
    _http_host = _environ["HTTP_HOST"]
    if _http_host.find("localhost") > -1:
      self._validate_certificate = False

    _wsgi_url_scheme = _environ["wsgi.url_scheme"]
    self._endpoint = {
      "JP":_wsgi_url_scheme + "://webservices.amazon.co.jp/onca/xml",
      "US":_wsgi_url_scheme + "://webservices.amazon.com/onca/xml",
      "FR":_wsgi_url_scheme + "://webservices.amazon.fr/onca/xml",
      "IN":_wsgi_url_scheme + "://webservices.amazon.in/onca/xml",
      "IT":_wsgi_url_scheme + "://webservices.amazon.it/onca/xml",
      "UK":_wsgi_url_scheme + "://webservices.amazon.co.uk/onca/xml"
    }

    if self._endpoint.has_key(locale):
      self._product_advertising_url = self._endpoint[locale]
    else:
      self._product_advertising_url = self._endpoint["US"]

    self._service = "AWSECommerceService"

    #self._method = ""
    #self._prms = {}

  #タイムスタンプを設定
  def _timestamp(self,prms):

    prms["Timestamp"] = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")

  #URLエンコード
  def _urlencode_rfc3986(self,s):
    _s = urllib.quote(s)
    _s = _s.replace("%7E", "~")
    return _s

  #アクセストークン用の文字列作成
  def _canonical_string(self,prms):

    _keys = sorted(prms)
    _canonical_string = ""
    for _k in _keys:
      _canonical_string = _canonical_string + "&" + self._urlencode_rfc3986(_k) + "=" + self._urlencode_rfc3986(prms[_k])
    else:
      _canonical_string = _canonical_string[1:]

    return _canonical_string

  #問い合わせURLの作成
  def _request_url(self,prms,method):

    _url_info = urlparse(self._product_advertising_url)

    _canonical_string = self._canonical_string(prms)

    _signature = method + "\n" + _url_info.hostname + "\n" + _url_info.path +"\n" + _canonical_string
    _signature = hmac.new(self._secret_access_key, _signature, hashlib.sha256).digest()
    _signature = base64.b64encode(_signature)

    _url = self._product_advertising_url + "?" + _canonical_string + "&Signature=" + self._urlencode_rfc3986(_signature)

    return _url

  #itemsearch
  def itemsearch(self,searchindex=None,browsenode=None,availability="Available",responsegroup="Small",itempage=1,keywords=None,minimumprice=None,maximumprice=None,isremoveadult=True,sort=None):

    _prms = {}

    _prms["Service"] = self._service
    _prms["AWSAccessKeyId"] = self._access_key_id

    if self._associate_tag is not None:
      _prms["AssociateTag"] = self._associate_tag

    _prms["Operation"] = "ItemSearch"
    self._timestamp(_prms)
    
    if availability == "":
      return None

    if availability is None:
      return None

    _prms["Availability"] = availability

    if searchindex == "":
      return None

    if searchindex is None:
      return None

    _prms["SearchIndex"] = searchindex

    if responsegroup is None:
      return None

    if responsegroup == "":
      return None

    _prms["ResponseGroup"] = responsegroup

    if browsenode is not None:
      _prms["BrowseNode"] = browsenode

    if itempage is not None:
      _prms["ItemPage"] = str(itempage)

    if keywords is not None:
      _prms["Keywords"] = keywords

    if maximumprice is not None:
      _prms["MaximumPrice"] = str(maximumprice)

    if minimumprice is not None:
      _prms["MinimumPrice"] = str(minimumprice)

    if sort is not None:
      _prms["Sort"] = sort

    _url = self._request_url(_prms,"GET")

    try:
      _result = urlfetch.fetch(_url,validate_certificate = self._validate_certificate)
      _status_code = _result.status_code
      #logging.info(_status_code)


      if _status_code == 200:

        _content = _result.content
        _root = etree.fromstring(_content)
        _ns = _root.xpath('namespace-uri(.)')
        _namespace = {"ns":_ns}
        _errorTags = _root.findall('.//ns:Error',namespaces=_namespace)
        #logging.info(_content)
        if len(_errorTags) > 0:
          _errorTag = _errorTags[0]
          _code  = _errorTag.findall('.//ns:Code',namespaces=_namespace)
          _message = _errorTag.findall('.//ns:Message',namespaces=_namespace)
          logging.error(u"code: " + _code[0].text + " message: " + _message[0].text)
        else:
          return _root
      else:
        logging.error(u"status code: " + str(_status_code) + " content:" + _content)
        
    except urlfetch.DeadlineExceededError:
      logging.error(u"urlfetch DeadlineExceededError")

    except urlfetch.ResponseTooLargeError, response:
      logging.error(u"urlfetch ResponseTooLargeError" + response)

    except urlfetch.InternalTransientError:
      logging.error(u"urlfetch InternalTransientError")

    except urlfetch.InvalidURLError:
      logging.error(u"urlfetch InvalidURLError")

    except urlfetch.Error:
      logging.error(u"urlfetch Error")

    return None

[Python]AmazonのProduct Advertising APIでkindle storeだけを検索する方法

AmazonのProduct Advertising APIでkindle storeのコンテンツだけをitem searchすることはできるのだろうか？

調査してみると、
SearchIndex = "KindleStore"
BrowseNode = "2250739051"
で設定して検索をかけるとヒットするみたいで、確かに実行してみたら、取得できた。

[Google App Engine][Python][SendGrid]メールを送信

SendGridを使ってメール送信する方法を調査しました。

import sendgrid
from sendgrid.helpers import mail

sg = sendgrid.SendGridAPIClient(apikey = "API_KEY")
to_email = mail.Email("TO_MAIL_ADDRESS")
from_email = mail.Email("FROM_MAIL_ADDRESS")
subject = u"タイトル"
content = mail.Content('text/plain', u"本文")
message = mail.Mail(from_email, subject, to_email, content)
response = sg.client.mail.send.post(request_body = message.get())
status_code = int(response.status_code)
if status_code == 200 or status_code == 202:
  #正常終了
else:
  #エラー

[Google App Engine][Python][sendgrid]プログラムの準備

Google App Engine上でSendGridを使うにあたってSDKを用意する必要があるのだが、

SendGridのpython-http-client
と
SendGridのsendgrid-python
をそれぞれダウンロードする必要性があり、かつ、python-http-clientは「python_http_client」とリネームをする必要があります。

ちょっと注意ですね。

[Google App Engine][Python][sendgrid]ステータスコード

Google App Engine上でSendGridを使ってメールを送信する場合、ステータスコードがどこにあるのか気になって検索をかけたら
Status Codes & Errors
にあった。

こういうのって処理を組む上でものすごく重要なんだよなー。

[Google App Engine][Python]アダルトコンテンツについて

Google App Engineってアダルトコンテンツのwebサービスって作ってもいいのだろうか？

検索をかけたらStackOverflowに答えがありました。
Does Google App Engine allow adult content?
なんと、OKみたいです。

まさかと思って、ポリシーを見てみると、確かにない気がする。
Google Cloud Platform Acceptable Use Policy

前はNGだったのに。といっても、6、7年前ぐらいのことだけど。

許可してしまえば、他のクラウドサービスとの差別化になるのは容易に想像つくんだけど、果たしてどれだけの人が気づいているのやら。

これは結構でかい動きのような気がする。

[Google App Engine][Python]cronをローカルで実行する

cronをローカルで実行するにはどうすればいいのだろうか？

Scheduling Tasks With Cron for Python

で確認すると、http://localhost:8000/cronにアクセスして表示された実行ボタンを押せばおｋ

[Python]strftimeで日本語を使いたい

strftimeメソッドで日本語を使いたい場合はどうすればいいのだろうか？

Python datetime
によると、

import datetime
now = datetime.datetime.now()
s = now.strftime(u'%Y年X%m月X%d日'.encode('utf-8')).decode('utf-8')
print s

でおｋみたい。

[Python]曜日を取得する

JavaScriptのように

var now = new Date();
now.getDay();

曜日を番号でとりたい場合どうすればいいのだろうか？

import pytz
tz = pytz.timezone('Japan')
now = datetime.datetime.now(tz)
weekday = now.weekday()

このようにすればweekdayはintで取得することができます。

参考
8.1. datetime — Basic date and time types
datetimeで曜日を調べる

[Google App Engine][Python][Jinja2]htmlエスケープをしない

通常、Jinja2を使ってhtmlを描画する際に、

import os
import jinja2

_template_path = os.path.dirname(__file__) + "/tmplt"
jinja2.Environment(
  loader = jinja2.FileSystemLoader(_template_path),
  extensions = ['jinja2.ext.autoescape'],
  autoescape = True
)

と書いてhtmlをエスケープさせるのですが、一部、エスケープさせずにそのまま実行させたい場合があります。

そんな場合は、safeを使うことで実行することができます。

{{ html | safe }}

[Google App Engine][Python]cron dispatchの関係

例えばcrom.yamlで下記のように設定していたとする。

cron:
- description: "this is cron"
  url: /hoge
  schedule: every 1 hours
  target: hoge
  timezone: Japan
  retry_parameters:
    job_retry_limit: 0

でかたやdispatch.yamlで

dispatch:
- url: '*/hoge/*'
  service: foo

となった場合、service:fooが優先されます。