U n/eɆ@sddlZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddlmZmZddlmZddlmZddlmZmZddlmZd d ZGd d d eZGd ddeZdS)N) itemgetter) bucket_nameBase)DatasetManager)check_http_status_code)check listDatanode)find_list_datanode_from_datecCs|ddkr|d7}|S)N/)txtr r 7/tmp/pip-unpacked-wheel-t2z2wkof/dsmlibrary/datanode.py append_slashs rc @seZdZdddZddZddd Zd d d Zd!d d ZddifddZd"ddZ d#ddZ d$ddZ d%ddZ d&e jdddZd'e jdddZdS)(DataNodeNFc Kst|tjtjfkr&tdt||dks:t|tkrLtdt||dks`t|tkrrtdt||dkst|tkrd|}|d}|dkr|||n|}tj |j d|j |||d |d d } | j d krtd | | } d|ko|d |d<d|kr<|dr<|ddddt|tjkr^dd|D}|jdtd| df|jdd|tj|j | d|j d} |r| rtj|j d| dd|j dtj|j d| dd|j dd| d| d d!} |dkrt|d"kr| rtj|d#d$tj |j d| dd%|j d#|id } zt| Wn8tk r}z| t|d d&W5d}~XYnX| d'dd(| S))ab _summary_ Args: df (dask or pandas DataFrame, required): _description_. Defaults to None. directory (int, required): _description_. Defaults to None. name (str, required): _description_. Defaults to None. description (str, optional): _description_. Defaults to "". replace (bool, optional): _description_. Defaults to None. profiling (bool, optional): _description_. Defaults to False. lineage (list, optional): _description_. Defaults to None. Returns: dict: status z]Invalid type expect ddf(dask dataframe) or df(pandas dataframe), Please input `df=`, but got N(Please input data `name`=, but got -Please input data `directory`=, but got rzdata .parquet/file/F)name description directoryis_usereplaceheadersjsonz&can not create directory in discovery append overwritecSsdtjdS)Nzpart-r)datetimenow isoformat)xr r rEz DataNode.write..T) name_functionignore_divisionscSsi|]\}}|dkr||qS))r!r r(r)r ).0kvr r r Jsz"DataNode.write..s3://r keyZpyarrow)storage_optionsZengineidZ discovery_apifile_idheaderz/createDatadict/rz/createProfileling/path)sucessr3r6rlineagedataZdataName /setLineage/Z lineage_msgr8createdr8r8)typepd DataFramedd Exceptionstrint_check_fileExistsrequestspost_discovery_api _jwt_header status_coderupdateitemsZ to_parquetr_storage_optionsrcheck_file_alreadygetlencheck_lists_intr)selfdfrrrr profilingr8kwargs_resmeta _file_already _responserer r rwrites~    ""zDataNode.writec Cs|}|d|} ||| } |dkr*| }| sftj|jd|j| ||d| d|||diidd} n|j| |d } tj|jd| d |jd } tj| d | } | d idi}| |||ditj |jd| d |jd d|iid}tj|d tj| d | } tj |j| d|jd}|dkrt |dkr|rtj|ddtj|jd| dd|jd|id}zt|Wn8tk r}z| t|ddW5d}~XYnX| dddd| d| ddS)N.rrTversion_file_id) timestampr3)rrrrrcontextr)r directory_idr r5responserar1r2rr8r9r;Fr<r=r>r7r6)statusr3r6)Z_check_fileExists_no_askrGrHrIrJ get_file_idrPrrrrLpatchrOrQrRrCrD)rSr? file_namer3rbr8rr`_file_idrZ_is_file_existsrWZlist_datanode_idrXr_Z_DataNode__resrYr[r\r r r_write_listfilevs    zDataNode._write_listfilec Csxtj|}|d\}}tjd}|d|d|} |j||| |dd} | dd} |j d|| |||d } | S) Nr^%Y/%m/%d %H:%M:%S-T)rb file_pathrhrrr1rZlistFile)r?rhr3rbr8r) osr6basenamesplitr"r#strftimeZ upload_filerPrj) rSrbrmrr8rZoriginal_file_nameZfile_extensionr`rhrZriresr r r writeListFiles*   zDataNode.writeListFilec  Ks|r0tj} tj| j| j| jd} | d} ntjd} | rdtj| j| j| jd}|d} |j|||d| ||||d}| dd}|j d|||||| d}|S) aQ_summary_ Args: df (_type_, optional): _description_. Defaults to None. directory_id (_type_, optional): _description_. Defaults to None. name (_type_, optional): _description_. Defaults to None. description (str, optional): _description_. Defaults to "". replace (_type_, optional): _description_. Defaults to None. profiling (bool, optional): _description_. Defaults to False. lineage (_type_, optional): _description_. Defaults to None. Returns: _type_: _description_ )yearmonthdayrkrl)rTrrrrrUr8r3r listDataNode)r?rhr3rbr8rr`) r"datetodayrtrurvrqr#r]rPrj)rSrTrbrrrrUr8Zoverwrite_same_date data_daterVryZtoday_with_timeZ datetime_strZdatetime_inputrZrirrr r rwriteListDataNodes8   "  zDataNode.writeListDataNodec Cstj|jd|d|jd}|jdkrL|jdkr:|nd}td||}z|jjt |dd }Wn&tk r}z|W5d }~XYn:X| d |d d id d| D}|t |jfSd S)z_summary_ Args: file_id (_type_, optional): _description_. Defaults to None. Raises: Exception: _description_ e: _description_ Returns: _type_: _description_ rr r5 Some thing wrong, s3_key)rZ object_nameNownerusercSsi|]\}}|dkr||qS))rrrr6rZ human_sizer )r*r/valuer r rr--sz%DataNode.get_file..)rGrPrIrJrKrrCclientZ get_objectrrLrMioBytesIOr:)rSr3rWrrXrdr\r r rget_files   zDataNode.get_filercCstj|jd|d|jd}t||}|dt|di|dd}|dkrtj d t d|dfd |j i|S|d krtj d t d|dfd |j i|S|d krtj|j d |dd|jd}t|tj|j d|dd|jd}t||}t|j|j|j |jd} |d} | j| d\} } |dd} td| d|d| dtdd|dD}| d}| ||dt|}dd|D}| j| t|d|d S|d!krtj|||j|j |j t d"Std#|d$S)%a8_summary_ Args: file_id (_type_, optional): _description_. Defaults to None. index (_type_, optional): _description_. Defaults to None. Raises: Exception: _description_ e: _description_ Returns: _type_: _description_ rr r5r/rr?rparquetr.r0csvz sql-queryz /api/v2/file/r1z /getSqlQuery//api/sql/query/)tokenapikeyZdataplatform_api_uriZobject_storage_uridatabase database_idwhere_conditionr  from sqlalchemy import sql query cSsi|]}|d|dqScolumnr?r r*itemr r rr-csz%DataNode.read_ddf..rXsqlalchemy_uri pk_columncSsi|]}|d|dqSrr rr r rr-fs)rquery_functionrrXrwrXindexrZbase_urir0Z bucketNameCan not read file extension , support [parquet, csv])rGrPrIrJrrrLrrB read_parquetrrNread_csv_base_discovery_apiDatabaseManagementrrZ_base_minio_urlget_database_schemaexecglobals_get_order_meta_datarread_sql_queryr get_list_datanoderC)rSr3rZ extra_paramrWrX_f_typer[r:Z _databaser database_metaschemarreordered_metacon_strZupdated_reordered_metar r rread_ddf1s\  (( "&    zDataNode.read_ddfcCstj|jd|d|jd}|jdkrL|jdkr:|nd}td||}|dt|d i|d d }|d krt j d t d|d|j dS|dkrt j d t d|d|j dStd|dS)z_summary_ Args: file_id (_type_, optional): _description_. Defaults to None. Raises: Exception: _description_ Returns: _type_: _description_ rr r5r|r}r~rr/rr?rrr.)r0rrr)rGrPrIrJrKrrCrLrr@rrrNr)rSr3rWrrXrr r rread_dfps      zDataNode.read_dfcCsVtj|dtdtj|dtdtj|jd|j||dd}t|| dS)z_summary_ Args: name (_type_, optional): _description_. Defaults to None. directory_id (_type_, optional): _description_. Defaults to None. Returns: _type_: _description_ rvariableZ variableNameZdtyperbz/file/get-file-id/)rrrr1) r check_typerDrErGrHrIrJrrrP)rSrrbr[r r rrfs zDataNode.get_file_idcCsZtj|dtdtj|dtdtj|jd|j||dd}tj|d| dS) z_summary_ Args: parent_dir_id (_type_, optional): _description_. Defaults to None. name (_type_, optional): _description_. Defaults to None. Returns: _type_: _description_ parent_dir_idrrz/directory/get_directory_id/) parent_dirrrrcr1) rrrErDrGrHrIrJrrrP)rSrrrrr r rget_directory_ids  zDataNode.get_directory_idcCsttj|dtdtj|jd|d|jd}tj|d|}|didi}| }t |t d d d }|S) _summary_ Args: file_id (_type_, optional): _description_. Defaults to None. Returns: _type_: _description_ r3rrr r5rcrar_r`F)r/reverse) rrrErGrPrIrJrrvaluessortedr)rSr3rWrXversion_file_idsr r rget_file_versions  zDataNode.get_file_version)rzcCstj|dtdtj|jd|d|jd}tj|d|}|didi}| }t ||\}}|d krt d |d |d ||fS) rr3rrr r5rcrar_Nz data_date z) is not exist in list data node (file_id=)) rrrErGrPrIrJrrrr ValueError)rSr3rzrWrXrr result_listr r rget_file_from_dates  zDataNode.get_file_from_datec Cs(tj|jd|d|jd}t||}|dt|di|dd}|dkrd }d }|r|j||d \}} |d kr|}|d }t j |||j|j |j t d} |d kr| Sz t j |||j|j |j t d} Wn6tk r } z|jrt| d} W5d} ~ XYnXt | | Std|dS)rrr r5r/rr?rrwr )r3rzrrrNrz, support only [listDataNode])rGrPrIrJrrrLrrr rrrNrrCZ_verboseprintZget_change_data) rSr3rzrWrXrZ target_indexZprevious_indexrrZ ddf_currentZ ddf_previousr\r r rget_update_datasR       zDataNode.get_update_data)NNNrNFN)NNrNN) NNNrNFNFN)N)N)NN)NN)N)NN)NN)__name__ __module__ __qualname__r]rjrsr{rrrrfrrr"rxrrr r r rrs, ^M  3 ?    rc@sfeZdZdddZdddZdddZdd Zd d Zd d ZddZ dddZ ddZ dddZ dS)rNcCst|tkr tdt|dtj|jd|d|jd}|jdkrN|jS|jdkr`| S| }d|krx|dSd |S) zLGet sqlalchemy connection string Args: ... zExpect `con_id`= but got z#, please input `con_id` eg con_id=0/api/sql/database/r r5r}rzSome thing wrong!, ) r?rErCrGrPrrJrKcontentr)rScon_idr[r:r r r_get_connection_strs   z&DatabaseManagement._get_connection_strcCst|tkr tdt|dtj|jd|d|jd}|jdkrR|jdfS|jdkrh| dfS| }| d }||fS) z_summary_ Args: table_id (_type_, optional): _description_. Defaults to None. Raises: Exception: _description_ Returns: _type_: _description_ z Expect `table_id`= but got z', please input `table_id` eg table_id=0z/api/sql/table/r r5irr schema_code) r?rErCrGrPrrJrKrrpop)rSZtable_idr[rXrr r rget_table_schema,s      z#DatabaseManagement.get_table_schemac Os|dkr@tj|jd|jd|id}t|d|dd}tj|dt dtj|jd |d |jd }tj|d|}| d }||fS) z_summary_ Args: database_id (_type_, optional): _description_. Defaults to None. Returns: _type_: _description_ N"/api/sql/database/get_database_id/rrrcr1rrrr r5r) rGrHrrJrrrPrrrEr) rSr database_nameargsrVrrr[rXrr r rrDs    z&DatabaseManagement.get_database_schemacCsdS)zbCheck table and column accesing permission from con_id Args: ... Nr )rSrtablerr r rcheck_permission\sz#DatabaseManagement.check_permissioncCsH|jD]<}|djj}|j|||ddstd|d|dqdS)zoCheck table and column accesing permission from sqlalchemy query Args: ... exprr)rrrz#You don't have permission in table=z , column=N)Zcolumn_descriptionsrrrrC)rSrrrZ table_namer r rcheck_query_permissiones  z)DatabaseManagement.check_query_permissioncCs@tjt|d}||dkr(d||<||}||}|S)N)columnsZInt64Zint64)r@rAlistkeysZastypeZ set_index)rSrXrZ meta_daskr r r_create_pandas_metars    z&DatabaseManagement._create_pandas_metac Cs|}||dkr*tj||||d}ntj|||dd}|j}zdg}|D]B} | |krp|| || dqNtd| d|| ddqN||||dWn0tk r} zt| d W5d} ~ XYnX|S) NrDstring)sqlcon index_col divisionsr)rrrZ npartitionsrzWarning! column "z" exist in data source but not exist in reflection schema, so it will be auto cast to "string". If you want to use data type same as data source, please regenreate reflection schemarz. This column is exist in data source but not exist in reflection schema. Maybe this is a new column in data source. Please check your reflection schema file and regenreate it)_get_possible_str_divisionrBrrr rKeyError) rSrXrrrpossible_divisionsddfZ columns_orderrrZker r rrzs   z'DatabaseManagement._get_order_meta_datac  Os|dks"t|dkr6t|tkr6tdt|d|dksJt|tkr^tdt|d|dksrt|tkrtdt|d|dkst|tkrtd t|d |dkst|tkrtd t|d |dkst|tkrtd t|d|dkrd|}|dkr(|||n|} |j|d\} } | d}t|r`t |}|}n0t|tkr|}t d| d|dt t }|j ||t |d}tj|jd|j|||||||| dd}t||jdkr|S|jS)a_summary_ Args: query_function (_type_, optional): _description_. Defaults to None. directory_id (_type_, optional): _description_. Defaults to None. database_id (_type_, optional): _description_. Defaults to None. pk_column (_type_, optional): _description_. Defaults to None. meta (_type_, optional): _description_. Defaults to None. name (_type_, optional): _description_. Defaults to None. desciption (_type_, optional): _description_. Defaults to None. replace (_type_, optional): _description_. Defaults to None. Raises: Exception: _description_ Exception: _description_ Exception: _description_ Exception: _description_ Exception: _description_ Exception: _description_ Returns: _type_: _description_ NFz(Expect `query`=/ but got , please input `query`rz eg directory_id=0z/Please input data `database_id`=, but got z eg database_id=0z*Please input data `con_id`=, but got z eg pk_column='id' z)Please input data `meta`=, but got z eg meta='{'id':'Int64'}' rz eg name='myData' z query of rrrrr)rXrrrr)rrrrrrrXrrr})inspect isfunctionr?rDrCrEdictrFr getsourcerrrrrGrHrrJrrKrr)rSrrbrrrXrZ desciptionrrrV_replacerrrZquery_function_strZquery_function_functionrrrr r rwrite_sql_querysb"     z"DatabaseManagement.write_sql_querycCs tdtjtj}||S)N 0123456789)rrascii_lowercaseascii_uppercasesort)rSrr r rrsz-DatabaseManagement._get_possible_str_divisioncOs|dkr@tj|jd|jd|id}t|d|dd}tj|dt dtj|d t dtj|d t d|dkst |d krtd t|d |j|d\} } | d} |j||d} ||dkr|} tj|| || | d}ntj|| || dd}|S)a%_summary_ Args: table_id (_type_, optional): _description_. Defaults to None. query_function (_type_, optional): _description_. Defaults to None. pk_column (_type_, optional): _description_. Defaults to None. meta (_type_, optional): _description_. Defaults to None. Raises: Exception: _description_ Exception: _description_ Exception: _description_ Exception: _description_ Returns: _type_: _description_ Nrrrrcr1rrrrXFz"Expect `query`= but got rrr)rXrr)rrrrXrz300 MB)rrrrXZbytes_per_chunk)rGrHrrJrrrPrrrErDrrrrCr?rrrrBr)rSrrrrXrrrVrrrrrZdf_metarrr r rrsB   z!DatabaseManagement.read_sql_query)N)N)NN)NNNNNNNN)NNNNN) rrrrrrrrrrrrrr r r rrs      dr)rtimerrrGrnr"Zpandasr@Zdask.dataframeZ dataframerBoperatorrbaserrZdatasetrZutils.requestsrutilsrr Zutils.listDatanoder rrrr r r rs*