KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use start_query method in localstack

Best Python code snippet using localstack_python

sql_queries.py

Source:sql_queries.py

1class PersonQuery:2    START_QUERY = """3                SELECT p.id, 4                       p.updated_at5                FROM content.person p6                """7    FILTER_QUERY = """ WHERE updated_at > '{}'"""8    END_QUERY = """ ORDER BY updated_at"""9    @staticmethod10    def get_query(updated_at):11        if updated_at:12            return PersonQuery.START_QUERY + PersonQuery.FILTER_QUERY.format(updated_at) + PersonQuery.END_QUERY13        return PersonQuery.START_QUERY + PersonQuery.END_QUERY14class GenreQuery:15    START_QUERY = """16        SELECT g.id, g.updated_at17        FROM content.genre g 18    """19    FILTER_QUERY = """ WHERE updated_at > '{}'"""20    END_QUERY = """ ORDER BY updated_at"""21    @staticmethod22    def get_query(updated_at):23        if updated_at:24            return GenreQuery.START_QUERY + GenreQuery.FILTER_QUERY.format(updated_at) + GenreQuery.END_QUERY25        return GenreQuery.START_QUERY + GenreQuery.END_QUERY26class FWBase:27    BASE_QUERY = """28            SELECT DISTINCT fw.id, fw.updated_at29            FROM content.film_work fw30            """31    END_QUERY = """ ORDER BY fw.updated_at;"""32    FILTER_UPDATE_AT = """ AND fw.updated_at > %s"""33    PERSON_JOIN = f"""{BASE_QUERY} 34                        LEFT JOIN content.person_film_work pfw ON pfw.film_work_id = fw.id 35                        WHERE pfw.person_id IN %s"""36    GENRE_JOIN = f"""{BASE_QUERY} 37                    LEFT JOIN content.genre_film_work gfw ON gfw.film_work_id = fw.id 38                    WHERE gfw.genre_id IN %s"""39    @staticmethod40    def get_query(updated_at, linked_table):41        if linked_table == PersonQuery.__name__:42            start_query = FWBase.PERSON_JOIN43            updated_at_filter = FWBase.FILTER_UPDATE_AT44        elif linked_table == GenreQuery.__name__:45            start_query = FWBase.GENRE_JOIN46            updated_at_filter = FWBase.FILTER_UPDATE_AT47        else:48            start_query = FWBase.BASE_QUERY49            updated_at_filter = """ WHERE fw.updated_at > %s"""50        if updated_at:51            return start_query + updated_at_filter + FWBase.END_QUERY52        return start_query + FWBase.END_QUERY53class AllDataQuery:54    START_QUERY = """55            SELECT DISTINCT fw.id id,56                   fw.title title,57                   fw.description description,58                   fw.rating imdb_rating,59                   fw.updated_at updated_at,60                   ARRAY_AGG(DISTINCT jsonb_build_object('name', g.name, 'id', g.id)) AS genre,61                   ARRAY_AGG(DISTINCT jsonb_build_object('id', p.id, 'name', p.full_name)) FILTER (WHERE pfw.role='director') director,62                   ARRAY_AGG(DISTINCT jsonb_build_object('id', p.id, 'name', p.full_name)) FILTER (WHERE pfw.role='actor') actors,63                   ARRAY_AGG(DISTINCT jsonb_build_object('id', p.id, 'name', p.full_name)) FILTER (WHERE pfw.role='writer') writers,64                   ARRAY_AGG(DISTINCT p.full_name) FILTER (WHERE pfw.role='actor') actors_names,65                   ARRAY_AGG(DISTINCT p.full_name) FILTER (WHERE pfw.role='director') directors_names,66                   ARRAY_AGG(DISTINCT p.full_name) FILTER (WHERE pfw.role='writer') writers_names67        FROM content.film_work fw68            LEFT JOIN content.genre_film_work gfw on fw.id=gfw.film_work_id 69            LEFT JOIN content.person_film_work pfw on fw.id=pfw.film_work_id70            LEFT JOIN content.genre g on gfw.genre_id=g.id71            LEFT JOIN content.person p on pfw.person_id=p.id72        WHERE fw.id in %s73        GROUP BY fw.id 74        ORDER BY fw.updated_at75    """76    FILTER_BY_FW = """ WHERE fw.id in %s"""77    FILTER_BY_UPDATED_AT = """WHERE updated_at > %s"""78    @staticmethod79    def get_query():...

app.py

Source:app.py

1import numpy as np2import sqlalchemy3from sqlalchemy.ext.automap import automap_base4from sqlalchemy.orm import Session5from sqlalchemy import create_engine, func6from flask import Flask, jsonify7#################################################8# Database Setup9#################################################10engine = create_engine("sqlite:///Resources/hawaii.sqlite")11# reflect an existing database into a new model12Base = automap_base()13# reflect the tables14Base.prepare(engine, reflect=True)15# Save reference to the table16Measurement = Base.classes.measurement17Station = Base.classes.station18########19session = Session(engine)20#################################################21# Flask Setup22#################################################23app = Flask(__name__)24# Flask Routes25#################################################26@app.route("/")27def Welcome():28    """List all available api routes."""29    return (30        f"Available Routes:<br/>"31        f"/api/v1.0/precipition<br/>"32        f"/api/v1.0/station <br/>"33        f"/api/v1.0/tobs </br>"34        f"/api/v1.0/<start></br>"35        f"/api/v1.0/<start>/<end></br>"36    )37@app.route("/api/v1.0/precipitation")38def precipitation():39    session= Session(engine)40    sel= [Measurement.date, Measurement.prcp]41    prcp_data = session.query(*sel).all()42    session.close()43 # * Convert the query results to a dictionary using `date` as the key and `prcp` as the value.44  # Return the JSON representation of your dictionary.45    precipitation = []46    for date, prcp in prcp_data:47        prcp_dict = {}48        prcp_dict["date"] = date49        prcp_dict["prcp"] = prcp50        precipitation.append(prcp_dict)51    return jsonify(precipitation)52@app.route("/api/v1.0/stations")53def stations():54    session = Session(engine)55   #Query all stations56    station_list = session.query(Station.station).\57    order_by(Station.station).all()58    59#convert tuples to normal list60    stations = list(np.ravel(station_list))61    return jsonify(stations)62    63@app.route("/api/v1.0/tobs")64def tobs():65    session = Session(engine)66    lateststr = session.query(Measuremnt.date).order_by(Measurement.date.desc()).first[0]67    latestdate= dt.datetime.strptime(lateststr, "%Y-%m-%d")68    querydate = dt.date(latestdate.year -1, latestdate.month, latest.day)69    sel = [Measurement.date, Measurement.tobs]70    queryresult = session.query(*sel).filter(Measurement.date>=querydate).all()71    tempertaure = list(np.ravel(queryresult))72        73    return jsonify(tempertaure)74@app.route("/api/v1.0/<start>")75def start_date(start):76    session = Session(engine)77    start_query = session.query(func.min(Measurement.tobs), func.avg(Measurement.tobs), func.max(Measurement.tobs)).\78    filter(Measurement.date >=start).all()79    #tempertaure = list(np.ravel(start_query))80    start_temp = {}81    start_temp["TMIN"] = start_query[0][0]82    start_temp["TAVG"] = start_query[0][1]83    start_temp["TMAX"] = start_query[0][2]84    85    return jsonify(start_temp)86@app.route("/api/v1.0/<start>/<end>")87def start_end(start,end):88    session = Session(engine)89    start_query = session.query(func.min(Measurement.tobs), func.avg(Measurement.tobs), func.max(Measurement.tobs)).\90    filter(Measurement.date >=start).filter(Measurement.date<=end).all()91    #tempertaure = list(np.ravel(start_query))92    start_temp = {}93    start_temp["TMIN"] = start_query[0][0]94    start_temp["TAVG"] = start_query[0][1]95    start_temp["TMAX"] = start_query[0][2]96    97    return jsonify(start_temp)98if __name__ =='__main__':99    app.run(debug=True)...

fields.py

Source:fields.py

1#* coding: utf-82from django.db import models3class YmapCoord(models.CharField):4    def __init__(self, start_query=u'Ð Ð¾ÑÑÐ¸Ñ', size_width=500, size_height=500, **kwargs):5        self.start_query, self.size_width, self.size_height = start_query, size_width, size_height6        super(YmapCoord, self).__init__(**kwargs)7    def formfield(self, **kwargs):8        if 'widget' in kwargs:9            kwargs['widget'] = kwargs['widget'](attrs={10                "data-start_query": self.start_query,11                "data-size_width": self.size_width,12                "data-size_height": self.size_height,13            })14        return super(YmapCoord, self).formfield(**kwargs)15    def south_field_triple(self):16        "Returns a suitable description of this field for South."17        # We'll just introspect the _actual_ field.18        try:19            from south.modelsinspector import introspector20            field_class = self.__class__.__module__ + "." + self.__class__.__name__21            args, kwargs = introspector(self)22            # That's our definition!23            kwargs.update({24                'start_query': repr(self.start_query),25                'size_width': repr(self.size_width),26                'size_height': repr(self.size_height),27            })28            return ('django_ymap.fields.YmapCoord', args, kwargs)29        except ImportError:30            pass31    def deconstruct(self):32        name, path, args, kwargs = super(YmapCoord, self).deconstruct()33        if "start_query" in kwargs:34            del kwargs["start_query"]35        if 'size_width' in kwargs:36            del kwargs["size_width"]37        if 'size_height' in kwargs:38            del kwargs["size_height"]39        return name, path, args, kwargs40try:41    from south.modelsinspector import add_introspection_rules42    add_introspection_rules([], ["^django_ymap\.fields\.YmapCoord"])43except ImportError:...

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.