def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assertTrue(isinstance(r, DateRange))
self.assertTrue(not r.isempty)
self.assertEqual(r.lower, date(2000, 1, 2))
self.assertEqual(r.upper, date(2012, 12, 31))
self.assertTrue(not r.lower_inf)
self.assertTrue(not r.upper_inf)
self.assertTrue(r.lower_inc)
self.assertTrue(not r.upper_inc)
python类DateRange()的实例源码
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 0o1, 0o1)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertTrue(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assertTrue(r1.lower_inc)
self.assertTrue(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertTrue(isinstance(r1, DateTimeRange))
self.assertTrue(r1.isempty)
ts1 = datetime(2000, 1, 1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000, 12, 31, 23, 59, 59, 999,
tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertTrue(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assertTrue(not r1.lower_inc)
self.assertTrue(r1.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000, 1, 2))
self.assertEqual(r.upper, date(2012, 12, 31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 01, 01)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assert_(r1.lower_inc)
self.assert_(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeRange))
self.assert_(r1.isempty)
ts1 = datetime(2000, 1, 1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000, 12, 31, 23, 59, 59, 999,
tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assertTrue(isinstance(r, DateRange))
self.assertTrue(not r.isempty)
self.assertEqual(r.lower, date(2000, 1, 2))
self.assertEqual(r.upper, date(2012, 12, 31))
self.assertTrue(not r.lower_inf)
self.assertTrue(not r.upper_inf)
self.assertTrue(r.lower_inc)
self.assertTrue(not r.upper_inc)
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 0o1, 0o1)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertTrue(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assertTrue(r1.lower_inc)
self.assertTrue(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertTrue(isinstance(r1, DateTimeRange))
self.assertTrue(r1.isempty)
ts1 = datetime(2000, 1, 1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000, 12, 31, 23, 59, 59, 999,
tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assertTrue(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assertTrue(not r1.lower_inc)
self.assertTrue(r1.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 01, 01)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assert_(r1.lower_inc)
self.assert_(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeRange))
self.assert_(r1.isempty)
ts1 = datetime(2000,1,1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000,12,31,23,59,59,999, tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 01, 01)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assert_(r1.lower_inc)
self.assert_(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeRange))
self.assert_(r1.isempty)
ts1 = datetime(2000,1,1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000,12,31,23,59,59,999, tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 01, 01)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assert_(r1.lower_inc)
self.assert_(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeRange))
self.assert_(r1.isempty)
ts1 = datetime(2000,1,1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000,12,31,23,59,59,999, tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 01, 01)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assert_(r1.lower_inc)
self.assert_(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeRange))
self.assert_(r1.isempty)
ts1 = datetime(2000,1,1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000,12,31,23,59,59,999, tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 01, 01)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assert_(r1.lower_inc)
self.assert_(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeRange))
self.assert_(r1.isempty)
ts1 = datetime(2000,1,1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000,12,31,23,59,59,999, tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 01, 01)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assert_(r1.lower_inc)
self.assert_(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeRange))
self.assert_(r1.isempty)
ts1 = datetime(2000,1,1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000,12,31,23,59,59,999, tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 01, 01)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assert_(r1.lower_inc)
self.assert_(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeRange))
self.assert_(r1.isempty)
ts1 = datetime(2000,1,1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000,12,31,23,59,59,999, tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000, 1, 2))
self.assertEqual(r.upper, date(2012, 12, 31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 01, 01)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assert_(r1.lower_inc)
self.assert_(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeRange))
self.assert_(r1.isempty)
ts1 = datetime(2000, 1, 1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000, 12, 31, 23, 59, 59, 999,
tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
def get_queryset(self,):
params = self.request.query_params
showNulls = params.get('showNulls', None)
sourceName = params.get('source_name', None)
filteredFeatures = Feature.objects.all()
if sourceName:
filteredFeatures = filteredFeatures.filter(source_name=sourceName)
defaultRange = 180
defaultStart = datetime.date.today() - datetime.timedelta(days=2)
defaultEnd = defaultStart + datetime.timedelta(days=defaultRange)
startDate = params.get('startDate', None)
endDate = params.get('endDate', None)
if startDate or endDate:
if startDate and endDate:
try:
queryDateRange = DateRange(lower=startDate, upper=endDate)
except DataError:
queryDateRange = DateRange(lower=defaultStart, upper=defaultEnd)
elif startDate:
calcEnd = parser.parse(startDate) + datetime.timedelta(days=3)
queryDateRange = DateRange(lower=startDate, upper=calcEnd)
else:
queryDateRange = DateRange(lower=defaultStart, upper=defaultEnd)
filteredFeatures = filteredFeatures.filter(canonical_daterange__overlap=queryDateRange)
elif not showNulls:
filteredFeatures = filteredFeatures.exclude(canonical_daterange=None).exclude(canonical_daterange__isempty=True)
print('featurecount', filteredFeatures.count())
return filteredFeatures
def get_queryset(self):
params = self.request.query_params
excludeStatuses = ['COMPLETED', 'COMPLETED', 'COMPLETE', 'DENIED', 'CANCELED']
minDist = int(params.get('distance', 100))
minDays = int(params.get('days', 14))
defaultStart = datetime.date.today() - datetime.timedelta(days=2)
defaultEnd = defaultStart + datetime.timedelta(days=180)
startDate = params.get('startDate', defaultStart.isoformat())
endDate = params.get('endDate', defaultEnd.isoformat())
queryDateRange = DateRange(lower=startDate, upper=endDate)
excludeDateRange = DateRange(lower='1800-01-01', upper='2014-12-31')
collisionGraph = cache.get('featureGraph')
featureIDs = set()
for u, v, d in collisionGraph.edges(data=True):
if d['daysApart'] <= minDays and d['distance'] <= minDist:
featureIDs = {u, v} | featureIDs
filteredFeatures = Feature.objects\
.filter(pk__in=featureIDs)\
.filter(canonical_daterange__overlap=queryDateRange)\
.exclude(canonical_daterange__overlap=excludeDateRange)\
.exclude(canonical_status__in=excludeStatuses)
print('featurecount', filteredFeatures.count())
return filteredFeatures
def get_queryset(self):
params = self.request.query_params
excludeStatuses = ['COMPLETED', 'COMPLETED', 'COMPLETE', 'DENIED', 'CANCELED']
minDist = params.get('distance', 100)
minDays = params.get('days', 14)
# queryDate = parser.parse(self.request.query_params['date']).date()
# startDate = queryDate - datetime.timedelta(days=minDays)
# endDate = queryDate + datetime.timedelta(days=minDays)
# queryDateRange = DateRange(startDate, endDate)
defaultStart = datetime.date.today() - datetime.timedelta(days=2)
defaultEnd = defaultStart + datetime.timedelta(days=365)
startDate = params.get('startDate', defaultStart.isoformat())
endDate = params.get('endDate', defaultEnd.isoformat())
queryDateRange = DateRange(lower=startDate, upper=endDate)
excludeDateRange = DateRange(lower='1800-01-01', upper='2014-12-31')
geoaddy = AddressGeocode.objects.using('geocoder').raw("SELECT g.rating, ST_X(g.geomout) AS lon, ST_Y(g.geomout) AS lat, pprint_addy(addy) AS address FROM geocode(%s) as g LIMIT 1", [self.request.query_params['address']])[0]
queryPoint = GEOSGeometry('POINT({} {})'.format(geoaddy.lon, geoaddy.lat))
filteredFeatures = Feature.objects\
.filter(canonical_daterange__overlap=queryDateRange)\
.filter(geom__distance_lte=(queryPoint, D(m=minDist)))\
.exclude(canonical_daterange__overlap=excludeDateRange)\
.exclude(canonical_status__in=excludeStatuses)
print('featurecount', filteredFeatures.count())
return filteredFeatures
test_types_extras.py 文件源码
项目：aws-lambda-redshift-copy
作者: christianhxc
项目源码
文件源码
阅读 19
收藏 0
点赞 0
评论 0
def test_cast_date(self):
from psycopg2.extras import DateRange
cur = self.conn.cursor()
cur.execute("select '(2000-01-01,2012-12-31)'::daterange")
r = cur.fetchone()[0]
self.assert_(isinstance(r, DateRange))
self.assert_(not r.isempty)
self.assertEqual(r.lower, date(2000,1,2))
self.assertEqual(r.upper, date(2012,12,31))
self.assert_(not r.lower_inf)
self.assert_(not r.upper_inf)
self.assert_(r.lower_inc)
self.assert_(not r.upper_inc)
test_types_extras.py 文件源码
项目：aws-lambda-redshift-copy
作者: christianhxc
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
def test_adapt_date_range(self):
from psycopg2.extras import DateRange, DateTimeRange, DateTimeTZRange
from psycopg2.tz import FixedOffsetTimezone
cur = self.conn.cursor()
d1 = date(2012, 01, 01)
d2 = date(2012, 12, 31)
r = DateRange(d1, d2)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateRange))
self.assertEqual(r1.lower, d1)
self.assertEqual(r1.upper, d2)
self.assert_(r1.lower_inc)
self.assert_(not r1.upper_inc)
r = DateTimeRange(empty=True)
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeRange))
self.assert_(r1.isempty)
ts1 = datetime(2000,1,1, tzinfo=FixedOffsetTimezone(600))
ts2 = datetime(2000,12,31,23,59,59,999, tzinfo=FixedOffsetTimezone(600))
r = DateTimeTZRange(ts1, ts2, '(]')
cur.execute("select %s", (r,))
r1 = cur.fetchone()[0]
self.assert_(isinstance(r1, DateTimeTZRange))
self.assertEqual(r1.lower, ts1)
self.assertEqual(r1.upper, ts2)
self.assert_(not r1.lower_inc)
self.assert_(r1.upper_inc)
