Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use sns_api method in localstack

Best Python code snippet using localstack_python

sns_api.py

Source:sns_api.py

1# coding=utf-82from google.appengine.ext import ndb3from flask import Blueprint, request4import tweepy5import facebook as fb6import os7from constants import TWITTER_CONSUMER_KEY, TWITTER_CONSUMER_SECRET, APIStatus8from errors import DataError9from api import *10from utilities import response, get_form11from entities.sns import Facebook, Twitter12from entities.user import User13sns_api = Blueprint('sns', __name__, url_prefix='/api/sns')14@sns_api.route('/auth/facebook/', methods=['POST'])15def auth_facebook():16    from main import app17    form = get_form(FacebookForm(request.form))18    uuid, access_token, expires, code = form.uuid.data, form.access_token.data, form.expires.data, form.code.data19    user_key = User.get(uuid).key20    facebook = Facebook.get(user_key)21    if not facebook:22        app.logger.debug('Create Facebook, uuid: %s, access_token: %s' % (uuid, access_token))23        Facebook(parent=user_key, access_token=access_token, expires=expires, code=code).put()24    elif facebook.access_token != access_token:25        app.logger.debug('Update Facebook, uuid: %s, access_token: %s' % (uuid, access_token))26        facebook.access_token = access_token27        facebook.expires = expires28        facebook.code = code29        facebook.put()30    return response()31@sns_api.route('/share/facebook/', methods=['POST'])32def share_facebook():33    from main import app34    form = get_form(ShareForm(request.form))35    uuid, message, album, picture = form.uuid.data, form.message.data, form.album.data, form.picture.data36    user = User.get(uuid)37    facebook = Facebook.get(user.key)38    graph = fb.GraphAPI(facebook.access_token)39    share_msg = u'ãç±çã¯ã¤ãºã%s facebook.com/nekyou.quiz' % message40    try:41        if album and picture:42            app.logger.debug('Share picture, uuid: %s, message: %s, picture: %d' % (uuid, message, picture))43            graph.put_photo(open(get_picture_path(album, picture)), share_msg)44        else:45            app.logger.debug('Share message, uuid: %s, message: %s' % (uuid, message))46            graph.put_object('me', 'feed', link='http://www.facebook.com/nekyou.quiz', message=message)47    except Exception as e:48        app.logger.error('Facebook share failed: %s' % str(e))49    return response()50@sns_api.route('/auth/twitter/', methods=['POST'])51def auth_twitter():52    from main import app53    form = get_form(TwitterForm(request.form))54    uuid, token, token_secret, code = form.uuid.data, form.token.data, form.token_secret.data, form.code.data55    user_key = User.get(uuid).key56    twitter = Twitter.get(user_key)57    if not twitter:58        app.logger.debug('Create Twitter, uuid: %s, token: %s' % (uuid, token))59        Twitter(parent=user_key, token=token, token_secret=token_secret, code=code).put()60    elif twitter.token != token:61        app.logger.debug('Update Twitter, uuid: %s, token: %s' % (uuid, token))62        twitter.token = token63        twitter.token_secret = token_secret64        twitter.code = code65        twitter.put()66    return response()67@sns_api.route('/share/twitter/', methods=['POST'])68def share_twitter():69    from main import app70    form = get_form(ShareForm(request.form))71    uuid, message, album, picture = form.uuid.data, form.message.data, form.album.data, form.picture.data72    user = User.get(uuid)73    twitter = Twitter.get(user.key)74    auth = tweepy.OAuthHandler(TWITTER_CONSUMER_KEY, TWITTER_CONSUMER_SECRET)75    auth.set_access_token(twitter.token, twitter.token_secret)76    api = tweepy.API(auth)77    share_msg = u'ãç±çã¯ã¤ãºã%s facebook.com/nekyou.quiz' % message78    try:79        if album and picture:80            app.logger.debug('Share picture, uuid: %s, message: %s, picture: %d' % (uuid, message, picture))81            api.update_with_media(get_picture_path(album, picture), share_msg)82        else:83            app.logger.debug('Share message, uuid: %s, message: %s' % (uuid, message))84            api.update_status(share_msg)85    except Exception as e:86        app.logger.error('Twitter share failed: %s' % str(e))87    return response()88def get_picture_path(album, picture):89    if album == 1:90        path = os.path.join(os.path.dirname(__file__), '../../static/img/album/default/%d.png' % picture)91    elif album == 2:92        path = os.path.join(os.path.dirname(__file__), '../../static/img/album/second/%d.png' % picture)93    else:94        raise DataError(APIStatus.DATA_INCORRECT, 'Unsupported album: %d' % album)95    return path96class FacebookForm(BaseForm):97    access_token = StringField('access_token', [validators.input_required()])98    expires = IntegerField('expires', [validators.input_required()])99    code = StringField('code', [validators.input_required()])100class TwitterForm(BaseForm):101    token = StringField('token', [validators.input_required()])102    token_secret = StringField('token_secret', [validators.input_required()])103    code = StringField('code', [validators.input_required()])104class ShareForm(BaseForm):105    message = StringField('message', [validators.input_required()])106    album = IntegerField('album', [validators.optional()])...

code.py

Source:code.py

1from os import getenv2from os.path import join, dirname3import boto34class ConnectToSNS:5    def __init__(6        self, 7        sns_endpoint, 8        aws_region, 9        aws_access_key_id, 10        aws_secret_access_key, 11        aws_account_id, 12        sns_env, 13        sns_api,14        type_queue = 'sns',15        protocol = 'https'16    ):17        self.ENDPOINT = sns_endpoint18        self.type_queue = type_queue19        self.protocol = protocol20        self.region_name = aws_region21        self.aws_access_key_id = aws_access_key_id22        self.aws_secret_access_key = aws_secret_access_key23        self.CLIENT = boto3.client( type_queue, aws_region, aws_access_key_id, aws_secret_access_key )24        self.TOPIC_ARN = "arn:aws:sns:{}:{}:{}-{}".format( aws_region, aws_account_id, sns_env, sns_api )25    def suscribe(self):26        message = 'default error'27        try:28            suscription = self.CLIENT.subscribe(29                TopicArn=self.TOPIC_ARN,30                Protocol=self.protocol,31                Endpoint=f"{self.ENDPOINT}/sns/suscription"32            )33            return ( True, suscription )34        except self.CLIENT.exceptions.NotFoundException as e:35            message = e.response.get('Error').get('Message')36        except Exception as e:37            message = e38        return ( False, message )39    def confirm_suscription(self, token):40        message = 'default error'41        try:42            confirm = self.CLIENT.confirm_subscription(43                TopicArn=self.TOPIC_ARN,44                Token=token45            )46            return ( True, confirm )47        except self.CLIENT.exceptions.NotFoundException as e:48            message = e.response.get('Error').get('Message')49        except Exception as e:50            message = e...

main.py

Source:main.py

1from flask import Flask2from flask import render_template3from api.user_api import user_api4from api.audit_api import audit_api5from api.sns_api import sns_api6from errors import ParameterError, DataError7from utilities import response8from constants import APIStatus9from batch.notification import apns_push10app = Flask(__name__)11app.config.from_pyfile('settings.cfg')12app.register_blueprint(user_api)13app.register_blueprint(audit_api)14app.register_blueprint(sns_api)15@app.route('/')16def index():17    return render_template('index.html')18@app.route('/policy')19def policy():20    return render_template('policy.html')21@app.route('/terms')22def terms():23    return render_template('terms.html')24@app.route('/api/version/')25def version():26    return app.config['VERSION']27@app.route('/api/notification/')28def notification():29    apns_push()30    return 'Push notification test'31@app.errorhandler(404)32def page_not_found(e):33    return 'Sorry, nothing at this URL.', 40434@app.errorhandler(ParameterError)35def handle_parameter_error(error):36    app.logger.warn('Parameter error: %s' % error.to_dict())37    return response(APIStatus.PARAMETER_ERROR)38@app.errorhandler(DataError)39def handle_data_error(error):40    app.logger.error(error.to_dict())...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.