How to use on_post method in localstack

Best Python code snippet using localstack_python

views.py

Source:views.py

...39from mcod.users.models import Token40User = get_user_model()41class LoginView(BaseView):42    @versioned43    def on_post(self, request, response, *args, **kwargs):44        self.handle_post(request, response, self.POST_1_0, *args, **kwargs)45    @on_post.version('1.0')46    def on_post(self, request, response, *args, **kwargs):47        self.handle(request, response, self.POST_1_0, *args, **kwargs)48    class POST_1_0(CreateHandler):49        database_model = get_user_model()50        deserializer_schema = Login()51        serializer_schema = LoginSerializer(many=False, )  # include_data=('datasets',))52        def _data(self, request, cleaned, *args, **kwargs):53            cleaned['email'] = cleaned['email'].lower()54            try:55                user = User.objects.get(email=cleaned['email'])56            except User.DoesNotExist:57                raise falcon.HTTPUnauthorized(58                    title='401 Unauthorized',59                    description=_('Invalid email or password'),60                    code='account_not_exist'61                )62            if user.state != 'active':63                if user.state not in settings.USER_STATE_LIST or user.state == 'deleted':64                    raise falcon.HTTPUnauthorized(65                        title='401 Unauthorized',66                        description=_('Account is not available'),67                        code='account_unavailable'68                    )69                if user.state in ('draft', 'blocked'):70                    raise falcon.HTTPUnauthorized(71                        title='401 Unauthorized',72                        description=_('Account is blocked'),73                        code='account_unavailable'74                    )75                if user.state == 'pending':76                    raise falcon.HTTPForbidden(77                        title='403 Forbidden',78                        description=_('Email address not confirmed'),79                        code='account_inactive'80                    )81            user = authenticate(request=request, **cleaned)82            if user is None:83                raise falcon.HTTPUnauthorized(84                    title='401 Unauthorized',85                    description=_('Invalid email or password'),86                    code='authorization_error'87                )88            if not hasattr(request, 'session'):89                request.session = session_store()90                request.META = {}91            login(request, user)92            request.session.save()93            user.token = get_auth_token(user.email, user.system_role, request.session.session_key)94            return user95class RegistrationView(BaseView):96    @versioned97    def on_post(self, request, response, *args, **kwargs):98        self.handle_post(request, response, self.POST_1_0, *args, **kwargs)99    @on_post.version('1.0')100    def on_post(self, request, response, *args, **kwargs):101        self.handle(request, response, self.POST_1_0, *args, **kwargs)102    class POST_1_0(CreateHandler):103        database_model = get_user_model()104        deserializer_schema = Registration()105        serializer_schema = RegistrationSerializer(many=False, )106        def _clean(self, request, *args, **kwargs):107            cleaned = super()._clean(request, *args, **kwargs)108            if User.objects.filter(email__iexact=cleaned['email']):109                raise falcon.HTTPForbidden(110                    title='403 Forbidden',111                    description=_('This e-mail is already used'),112                    code='emial_already_used'113                )114            cleaned['email'] = cleaned['email'].lower()115            return cleaned116        def _data(self, request, cleaned, *args, **kwargs):117            usr = User.objects.create_user(**cleaned)118            token = usr.email_validation_token119            validation_path = settings.EMAIL_VALIDATION_PATH % token120            link = "%s%s" % (settings.BASE_URL, validation_path)121            # TODO: this is specific for mcod-backend, we should implement more generic solution122            try:123                conn = get_connection(settings.EMAIL_BACKEND)124                msg_plain = render_to_string('mails/confirm-registration.txt',125                                             {'link': link, 'host': settings.BASE_URL})126                msg_html = render_to_string('mails/confirm-registration.html',127                                            {'link': link, 'host': settings.BASE_URL})128                send_mail(129                    'Aktywacja konta',130                    msg_plain,131                    config.ACCOUNTS_EMAIL,132                    [usr.email, ],133                    connection=conn,134                    html_message=msg_html,135                )136            except SMTPException:137                raise falcon.HTTPInternalServerError(138                    description=_('Email cannot be send'),139                    code='email_send_error'140                )141            return usr142class AccountView(BaseView):143    @falcon.before(login_required)144    @versioned145    def on_get(self, request, response, *args, **kwargs):146        self.handle(request, response, self.GET_1_0, *args, **kwargs)147    @falcon.before(login_required)148    @on_get.version('1.0')149    def on_get(self, request, response, *args, **kwargs):150        self.handle(request, response, self.GET_1_0, *args, **kwargs)151    @falcon.before(login_required)152    @versioned153    def on_put(self, request, response, *args, **kwargs):154        self.handle(request, response, self.PUT_1_0, *args, **kwargs)155    @falcon.before(login_required)156    @on_put.version('1.0')157    def on_put(self, request, response, *args, **kwargs):158        self.handle(request, response, self.PUT_1_0, *args, **kwargs)159    class GET_1_0(RetrieveOneHandler):160        database_model = get_user_model()161        serializer_schema = RegistrationSerializer(many=False, )162        def _clean(self, request, *args, **kwargs):163            return request.user164    class PUT_1_0(UpdateHandler):165        database_model = get_user_model()166        deserializer_schema = AccountUpdate()167        serializer_schema = UserSerializer(many=False, )168class LogoutView(BaseView):169    @falcon.before(login_required)170    @versioned171    def on_post(self, request, response, *args, **kwargs):172        self.handle_post(request, response, self.POST_1_0, *args, **kwargs)173    @falcon.before(login_required)174    @on_post.version('1.0')175    def on_post(self, request, response, *args, **kwargs):176        self.handle(request, response, self.POST_1_0, *args, **kwargs)177    class POST_1_0(CreateHandler):178        database_model = get_user_model()179        def _clean(self, request, *args, **kwargs):180            return request.user181        def _data(self, request, cleaned, *args, **kwargs):182            logout(request)183        def _serialize(self, data, meta, links=None, *args, **kwargs):184            return {}185class ResetPasswordView(BaseView):186    @versioned187    def on_post(self, request, response, *args, **kwargs):188        self.handle_post(request, response, self.POST_1_0, *args, **kwargs)189    @on_post.version('1.0')190    def on_post(self, request, response, *args, **kwargs):191        self.handle(request, response, self.POST_1_0, *args, **kwargs)192    class POST_1_0(CreateHandler):193        database_model = get_user_model()194        deserializer_schema = PasswordReset()195        def _data(self, request, cleaned, *args, **kwargs):196            try:197                usr = User.objects.get(email=cleaned['email'])198            except User.DoesNotExist:199                raise falcon.HTTPNotFound(200                    description=_('Account not found'),201                    code='account_not_found'202                )203            reset_token = usr.password_reset_token204            reset_path = settings.PASSWORD_RESET_PATH % reset_token205            link = "%s%s" % (settings.BASE_URL, reset_path)206            # TODO: this is specific for mcod-backend, we should implement more generic solution207            try:208                conn = get_connection(settings.EMAIL_BACKEND)209                msg_plain = render_to_string('mails/password-reset.txt', {'link': link, 'host': settings.BASE_URL})210                msg_html = render_to_string('mails/password-reset.html', {'link': link, 'host': settings.BASE_URL})211                send_mail(212                    'Reset hasÅa',213                    msg_plain,214                    config.ACCOUNTS_EMAIL,215                    [usr.email, ],216                    connection=conn,217                    html_message=msg_html218                )219            except SMTPException:220                raise falcon.HTTPInternalServerError(221                    description=_('Email cannot be send'),222                    code='email_send_error'223                )224            return usr225        def _serialize(self, data, meta, links=None, *args, **kwargs):226            return {'result': 'ok'}227class ConfirmResetPasswordView(BaseView):228    @versioned229    def on_post(self, request, response, *args, **kwargs):230        self.handle_post(request, response, self.POST_1_0, *args, **kwargs)231    @on_post.version('1.0')232    def on_post(self, request, response, *args, **kwargs):233        self.handle(request, response, self.POST_1_0, *args, **kwargs)234    class POST_1_0(CreateHandler):235        database_model = get_user_model()236        deserializer_schema = PasswordResetConfirm()237        def _data(self, request, cleaned, token, *args, **kwargs):238            try:239                token = Token.objects.get(token=token, token_type=1)240            except Token.DoesNotExist:241                raise falcon.HTTPNotFound()242            if not token.is_valid:243                raise falcon.HTTPBadRequest(244                    description=_('Expired token'),245                    code='expired_token'246                )247            token.user.set_password(cleaned['new_password1'])248            token.user.save()249            token.invalidate()250            return token.user251        def _serialize(self, data, meta, links=None, *args, **kwargs):252            return {}253class ChangePasswordView(BaseView):254    @falcon.before(login_required)255    @versioned256    def on_post(self, request, response, *args, **kwargs):257        self.handle_post(request, response, self.POST_1_0, *args, **kwargs)258    @falcon.before(login_required)259    @on_post.version('1.0')260    def on_post(self, request, response, *args, **kwargs):261        self.handle(request, response, self.POST_1_0, *args, **kwargs)262    class POST_1_0(CreateHandler):263        database_model = get_user_model()264        deserializer_schema = PasswordChange()265        serializer_schema = UserSerializer(many=False, )266        def _clean(self, request, *args, **kwargs):267            cleaned = super()._clean(request, *args, **kwargs)268            usr = getattr(request, 'user', None)269            is_valid = usr.check_password(cleaned['old_password'])270            if not is_valid:271                raise falcon.HTTPUnprocessableEntity(272                    description=_('Wrong password'),273                )274            try:275                dj_validate_password(cleaned['new_password1'])276            except DjangoValidationError as e:277                raise falcon.HTTPUnprocessableEntity(278                    description=e.error_list[0].message,279                )280            if cleaned['new_password1'] != cleaned['new_password2']:281                raise falcon.HTTPUnprocessableEntity(282                    description=_('Passwords not match'),283                )284            return cleaned285        def _data(self, request, cleaned, *args, **kwargs):286            request.user.set_password(cleaned['new_password1'])287            request.user.save()288        def _serialize(self, data, meta, links=None, *args, **kwargs):289            return {}290class VerifyEmailView(BaseView):291    @versioned292    def on_get(self, request, response, *args, **kwargs):293        self.handle(request, response, self.GET_1_0, *args, **kwargs)294    @on_get.version('1.0')295    def on_get(self, request, response, *args, **kwargs):296        self.handle(request, response, self.GET_1_0, *args, **kwargs)297    class GET_1_0(RetrieveOneHandler):298        database_model = get_user_model()299        def _clean(self, request, token, *args, **kwargs):300            try:301                token = Token.objects.get(token=token, token_type=0)302            except Token.DoesNotExist:303                raise falcon.HTTPNotFound()304            if not token.is_valid:305                raise falcon.HTTPBadRequest(306                    description=_('Expired token'),307                    code='expired_token'308                )309            if not token.user.email_confirmed:310                token.user.state = 'active' if token.user.state == 'pending' else token.user.state311                token.user.email_confirmed = timezone.now()312                token.user.save()313                token.invalidate()314        def _data(self, request, cleaned, *args, **kwargs):315            return {}316        def _serialize(self, data, meta, links=None, *args, **kwargs):317            return {}318class ResendActivationEmailView(BaseView):319    @versioned320    def on_post(self, request, response, *args, **kwargs):321        self.handle_post(request, response, self.POST_1_0, *args, **kwargs)322    @on_post.version('1.0')323    def on_post(self, request, response, *args, **kwargs):324        self.handle(request, response, self.POST_1_0, *args, **kwargs)325    class POST_1_0(CreateHandler):326        database_model = get_user_model()327        deserializer_schema = ResendActivationEmail()328        def _clean(self, request, *args, **kwargs):329            cleaned = super()._clean(request, *args, **kwargs)330            try:331                usr = User.objects.get(email=cleaned['email'])332            except User.DoesNotExist:333                raise falcon.HTTPNotFound(334                    description=_('Account not found'),335                    code='account_not_found'336                )337            return usr...

app.py

Source:app.py

...5from celery_app import delay_one_min6class UserCreate(object):7    def __init__(self):8        self.conn = Client()9    def on_post(self,req,resp):10        data = req.stream.read()11        data = json.loads(data)12        username = data["username"]13        password = data["password"]14        response = self.conn.register(username,password)15        print("in post",response)16        resp.body = json.dumps({"token":response})17        #self.conn.authenticate(username,password)18class UserAuth(object):19    def __init__(self):20        self.conn = Client()21    def on_post(self,req,resp):22        data =req.stream.read()23        data =json.loads(data)24        username = data["username"]25        password = data["password"]26        result = self.conn.authenticate(username,password)27        if result == True:28            resp.body = json.dumps({"result":"Auth Success","status":"success"})29            return30        else:31            resp.body = json.dumps({"result":"Auth Failed","status":"failed"})32            return33class Hook(object):34    def __init__(self):35        self.client = hookClient.Client()36    def on_post(self,req,resp):37        print("in hook client")38        data =req.stream.read()39        data =json.loads(data)40        username = data["username"]41        callback = data["callback"]42        print("save hook")43        self.client.create(username,callback)44        print("before running the job")45        random_job = delay_one_min.delay(username)46        print("Job is initiated")47        # code to save the hook48        resp.body = json.dumps({"result":"callback registered. You'll recieive it once the random job is completed","status":"success"})49app = falcon.API()50user_create = UserCreate()...

controllers.py

Source:controllers.py

...6        resp.content = api.template('index.html')7class Segmenter():8    async def on_get(self, req, resp):9        resp.content = api.template('segmenter.html')10    async def on_post(self, req, resp):11        data = await req.media()12        jsonData = models.segmenter(data)13        resp.media = jsonData14class Similarity_calculator():15    async def on_get(self, req, resp):16        resp.content = api.template('similarity_calculator.html')17    async def on_post(self, req, resp):18        data = await req.media()19        jsonData = models.similarity_calculator(data)20        resp.media = jsonData21class Parser():22    async def on_get(self, req, resp):23        resp.content = api.template('parser.html')24    async def on_post(self, req, resp):25        data = await req.media()26        jsonData = models.parser(data)27        resp.media = jsonData28class Extract_evaluation():29    async def on_get(self, req, resp):30        resp.content = api.template('extract_evaluation.html')31    async def on_post(self, req, resp):32        data = await req.media()33        jsonData = models.extract_evaluation(data)...

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.