1212# See the License for the specific language governing permissions and
1313# limitations under the License.
1414
15+ import base64
16+ import datetime
17+ import decimal
1518import hashlib
19+ import json
1620import logging
1721import math
1822import pathlib
1923import textwrap
2024import traceback
2125import typing
22- from datetime import datetime
2326from typing import Dict , Generator , Optional
2427
28+ import db_dtypes # type: ignore[import-untyped]
2529import fsspec # type: ignore[import-untyped]
2630import gcsfs # type: ignore[import-untyped]
31+ import geopandas as gpd # type: ignore[import-untyped]
2732import google .api_core .exceptions
2833import google .cloud .bigquery as bigquery
2934import google .cloud .bigquery_connection_v1 as bigquery_connection_v1
3439import google .cloud .storage as storage # type: ignore
3540import numpy as np
3641import pandas as pd
42+ import pandas .arrays
3743import pyarrow as pa
3844import pytest
3945import pytz
@@ -496,14 +502,213 @@ def nested_structs_df(
496502
497503@pytest .fixture (scope = "session" )
498504def nested_structs_pandas_df (nested_structs_pandas_type : pd .ArrowDtype ) -> pd .DataFrame :
499- """pd.DataFrame pointing at test data."""
505+ """pd.DataFrame pointing at test data.
506+
507+ Manually parses using json.loads to preserve data types.
508+ """
509+ with open (DATA_DIR / "nested_structs.jsonl" ) as f :
510+ raw_rows = [json .loads (line ) for line in f ]
511+
512+ ids = [row ["id" ] for row in raw_rows ]
513+
514+ def get_val (row , col_name ):
515+ return row .get (col_name )
516+
517+ # person
518+ person_struct_schema = nested_structs_pandas_type .pyarrow_dtype
519+ processed_person : list [Optional [dict [str , typing .Any ]]] = []
520+ for row in raw_rows :
521+ x = get_val (row , "person" )
522+ if x is None :
523+ processed_person .append (None )
524+ else :
525+ d = dict (x )
526+ if "age" in d and d ["age" ] is not None :
527+ d ["age" ] = int (d ["age" ])
528+ processed_person .append (d )
529+ person_arr = pa .array (processed_person , type = person_struct_schema )
530+ person_ser = pd .Series (person_arr , index = ids , dtype = nested_structs_pandas_type )
531+
532+ # bool_col
533+ bool_vals = [
534+ bool (get_val (row , "bool_col" )) if get_val (row , "bool_col" ) is not None else None
535+ for row in raw_rows
536+ ]
537+ bool_ser = pd .Series (bool_vals , index = ids , dtype = pd .BooleanDtype ())
538+
539+ # int64_col
540+ int64_vals = [
541+ int (get_val (row , "int64_col" ))
542+ if get_val (row , "int64_col" ) is not None
543+ else None
544+ for row in raw_rows
545+ ]
546+ int64_ser = pd .Series (int64_vals , index = ids , dtype = pd .Int64Dtype ())
547+
548+ # float64_col
549+ float64_vals = [
550+ float (get_val (row , "float64_col" ))
551+ if get_val (row , "float64_col" ) is not None
552+ else None
553+ for row in raw_rows
554+ ]
555+ np_vals = np .array (
556+ [x if x is not None else np .nan for x in float64_vals ], dtype = np .float64
557+ )
558+ mask = np .array ([x is None for x in float64_vals ], dtype = bool )
559+ float64_arr = pd .arrays .FloatingArray (np_vals , mask ) # type: ignore
560+ float64_ser = pd .Series (float64_arr , index = ids )
561+
562+ # string_col
563+ string_vals = [
564+ str (get_val (row , "string_col" ))
565+ if get_val (row , "string_col" ) is not None
566+ else None
567+ for row in raw_rows
568+ ]
569+ string_ser = pd .Series (
570+ string_vals , index = ids , dtype = pd .StringDtype (storage = "pyarrow" )
571+ )
500572
501- df = pd .read_json (
502- DATA_DIR / "nested_structs.jsonl" ,
503- lines = True ,
573+ # json_col
574+ json_strs : list [Optional [str ]] = []
575+ for row in raw_rows :
576+ if "json_col" not in row :
577+ json_strs .append (None )
578+ elif row ["json_col" ] is None :
579+ json_strs .append ("null" )
580+ else :
581+ json_strs .append (
582+ json .dumps (row ["json_col" ], sort_keys = True , separators = ("," , ":" ))
583+ )
584+ json_arr = pa .array (json_strs , type = db_dtypes .JSONArrowType ())
585+ json_ser = pd .Series (
586+ json_arr , index = ids , dtype = pd .ArrowDtype (db_dtypes .JSONArrowType ())
587+ )
588+
589+ # date_col
590+ date_vals = [
591+ datetime .date .fromisoformat (get_val (row , "date_col" ))
592+ if get_val (row , "date_col" ) is not None
593+ else None
594+ for row in raw_rows
595+ ]
596+ date_arr = pa .array (date_vals , type = pa .date32 ())
597+ date_ser = pd .Series (date_arr , index = ids , dtype = pd .ArrowDtype (pa .date32 ()))
598+
599+ # time_col
600+ time_vals = [
601+ datetime .time .fromisoformat (get_val (row , "time_col" ))
602+ if get_val (row , "time_col" ) is not None
603+ else None
604+ for row in raw_rows
605+ ]
606+ time_arr = pa .array (time_vals , type = pa .time64 ("us" ))
607+ time_ser = pd .Series (time_arr , index = ids , dtype = pd .ArrowDtype (pa .time64 ("us" )))
608+
609+ # datetime_col
610+ datetime_vals : list [Optional [datetime .datetime ]] = []
611+ for row in raw_rows :
612+ val = get_val (row , "datetime_col" )
613+ if val is None :
614+ datetime_vals .append (None )
615+ else :
616+ datetime_vals .append (datetime .datetime .fromisoformat (val .replace (" " , "T" )))
617+ datetime_arr = pa .array (datetime_vals , type = pa .timestamp ("us" ))
618+ datetime_ser = pd .Series (
619+ datetime_arr , index = ids , dtype = pd .ArrowDtype (pa .timestamp ("us" ))
504620 )
505- df = df .set_index ("id" )
506- df ["person" ] = df ["person" ].astype (nested_structs_pandas_type )
621+
622+ # timestamp_col
623+ timestamp_vals = [
624+ datetime .datetime .fromisoformat (
625+ get_val (row , "timestamp_col" ).replace ("Z" , "+00:00" )
626+ )
627+ if get_val (row , "timestamp_col" ) is not None
628+ else None
629+ for row in raw_rows
630+ ]
631+ timestamp_arr = pa .array (timestamp_vals , type = pa .timestamp ("us" , tz = "UTC" ))
632+ timestamp_ser = pd .Series (
633+ timestamp_arr , index = ids , dtype = pd .ArrowDtype (pa .timestamp ("us" , tz = "UTC" ))
634+ )
635+
636+ # bytes_col
637+ bytes_vals : list [Optional [bytes ]] = []
638+ for row in raw_rows :
639+ val = get_val (row , "bytes_col" )
640+ if val is None :
641+ bytes_vals .append (None )
642+ elif val == "" :
643+ bytes_vals .append (b"" )
644+ else :
645+ bytes_vals .append (base64 .b64decode (val ))
646+ bytes_arr = pa .array (bytes_vals , type = pa .binary ())
647+ bytes_ser = pd .Series (bytes_arr , index = ids , dtype = pd .ArrowDtype (pa .binary ()))
648+
649+ # numeric_col
650+ numeric_vals = [
651+ decimal .Decimal (str (get_val (row , "numeric_col" )))
652+ if get_val (row , "numeric_col" ) is not None
653+ else None
654+ for row in raw_rows
655+ ]
656+ numeric_arr = pa .array (numeric_vals , type = pa .decimal128 (38 , 9 ))
657+ numeric_ser = pd .Series (
658+ numeric_arr , index = ids , dtype = pd .ArrowDtype (pa .decimal128 (38 , 9 ))
659+ )
660+
661+ # bignumeric_col
662+ bignumeric_vals = [
663+ decimal .Decimal (str (get_val (row , "bignumeric_col" )))
664+ if get_val (row , "bignumeric_col" ) is not None
665+ else None
666+ for row in raw_rows
667+ ]
668+ bignumeric_arr = pa .array (bignumeric_vals , type = pa .decimal256 (76 , 38 ))
669+ bignumeric_ser = pd .Series (
670+ bignumeric_arr , index = ids , dtype = pd .ArrowDtype (pa .decimal256 (76 , 38 ))
671+ )
672+
673+ # geography_col
674+ geo_vals = [get_val (row , "geography_col" ) for row in raw_rows ]
675+ geo_ser = gpd .GeoSeries .from_wkt (geo_vals )
676+ geo_ser .index = ids
677+
678+ # duration_col
679+ duration_vals = [
680+ int (get_val (row , "duration_col" ))
681+ if get_val (row , "duration_col" ) is not None
682+ else None
683+ for row in raw_rows
684+ ]
685+ duration_arr = pa .array (duration_vals , type = pa .duration ("us" ))
686+ duration_ser = pd .Series (
687+ duration_arr , index = ids , dtype = pd .ArrowDtype (pa .duration ("us" ))
688+ )
689+
690+ df = pd .DataFrame (
691+ {
692+ "person" : person_ser ,
693+ "bool_col" : bool_ser ,
694+ "int64_col" : int64_ser ,
695+ "float64_col" : float64_ser ,
696+ "string_col" : string_ser ,
697+ "json_col" : json_ser ,
698+ "date_col" : date_ser ,
699+ "time_col" : time_ser ,
700+ "datetime_col" : datetime_ser ,
701+ "timestamp_col" : timestamp_ser ,
702+ "bytes_col" : bytes_ser ,
703+ "numeric_col" : numeric_ser ,
704+ "bignumeric_col" : bignumeric_ser ,
705+ "geography_col" : geo_ser ,
706+ "duration_col" : duration_ser ,
707+ },
708+ index = ids ,
709+ )
710+ df .index .name = "id"
711+
507712 return df
508713
509714
@@ -834,9 +1039,9 @@ def new_time_series_pandas_df():
8341039 return pd .DataFrame (
8351040 {
8361041 "parsed_date" : [
837- datetime (2017 , 8 , 2 , tzinfo = utc ),
838- datetime (2017 , 8 , 3 , tzinfo = utc ),
839- datetime (2017 , 8 , 4 , tzinfo = utc ),
1042+ datetime . datetime (2017 , 8 , 2 , tzinfo = utc ),
1043+ datetime . datetime (2017 , 8 , 3 , tzinfo = utc ),
1044+ datetime . datetime (2017 , 8 , 4 , tzinfo = utc ),
8401045 ],
8411046 "total_visits" : [2500 , 2500 , 2500 ],
8421047 }
@@ -855,12 +1060,12 @@ def new_time_series_pandas_df_w_id():
8551060 return pd .DataFrame (
8561061 {
8571062 "parsed_date" : [
858- datetime (2017 , 8 , 2 , tzinfo = utc ),
859- datetime (2017 , 8 , 2 , tzinfo = utc ),
860- datetime (2017 , 8 , 3 , tzinfo = utc ),
861- datetime (2017 , 8 , 3 , tzinfo = utc ),
862- datetime (2017 , 8 , 4 , tzinfo = utc ),
863- datetime (2017 , 8 , 4 , tzinfo = utc ),
1063+ datetime . datetime (2017 , 8 , 2 , tzinfo = utc ),
1064+ datetime . datetime (2017 , 8 , 2 , tzinfo = utc ),
1065+ datetime . datetime (2017 , 8 , 3 , tzinfo = utc ),
1066+ datetime . datetime (2017 , 8 , 3 , tzinfo = utc ),
1067+ datetime . datetime (2017 , 8 , 4 , tzinfo = utc ),
1068+ datetime . datetime (2017 , 8 , 4 , tzinfo = utc ),
8641069 ],
8651070 "id" : ["1" , "2" , "1" , "2" , "1" , "2" ],
8661071 "total_visits" : [2500 , 2500 , 2500 , 2500 , 2500 , 2500 ],
@@ -1473,7 +1678,7 @@ def cleanup_cloud_functions(session, cloudfunctions_client, dataset_id_permanent
14731678 continue
14741679
14751680 # Ignore the functions less than one day old
1476- age = datetime .now () - datetime .fromtimestamp (
1681+ age = datetime .datetime . now () - datetime . datetime .fromtimestamp (
14771682 cloud_function .update_time .timestamp ()
14781683 )
14791684 if age .days <= 0 :
0 commit comments