o ;Di O@sbddlmZmZmZddlZddlZddlmZddlmZm Z m Z ddl m Z ddl mZmZmZmZddlmZmZddlmZmZdd lmZmZmZmZmZd2d d Z d3ddZd4ddZ GdddeZ! d5ddZ"ddZ#ddZ$ddZ%ddZ&ddZ'd d!Z(d"d#Z)d$d%Z*d&d'Z+d(d)Z,d*d+Z-d,d-Z.d.d/Z/d0d1Z0ee_e e_ dS)6)absolute_importdivisionprint_functionN) OrderedDict)datetimedatetime)Decimal) izip_longest text_type string_typesPY3)read_source_from_argwrite_source_from_arg)skip setheader)Tabledicts fieldnamesiterpeekwrapcKst|}td|||d|S)a Extract a table from the records of a avro file. The `source` argument (string or file-like or fastavro.reader) can either be the path of the file, a file-like input stream or a instance from fastavro.reader. The `limit` and `skip` arguments can be used to limit the range of rows to extract. The `sample` argument (int, optional) defines how many rows are inspected for discovering the field types and building a schema for the avro file when the `schema` argument is not passed. The rows fields read from file can have scalar values like int, string, float, datetime, date and decimal but can also have compound types like enum, :ref:`array `, map, union and record. The fields types can also have recursive structures defined in :ref:`complex schemas `. Also types with :ref:`logical types ` types are read and translated to coresponding python types: long timestamp-millis and long timestamp-micros: datetime.datetime, int date: datetime.date, bytes decimal and fixed decimal: Decimal, int time-millis and long time-micros: datetime.time. Example usage for reading files:: >>> # set up a Avro file to demonstrate with ... >>> schema1 = { ... 'doc': 'Some people records.', ... 'name': 'People', ... 'namespace': 'test', ... 'type': 'record', ... 'fields': [ ... {'name': 'name', 'type': 'string'}, ... {'name': 'friends', 'type': 'int'}, ... {'name': 'age', 'type': 'int'}, ... ] ... } ... >>> records1 = [ ... {'name': 'Bob', 'friends': 42, 'age': 33}, ... {'name': 'Jim', 'friends': 13, 'age': 69}, ... {'name': 'Joe', 'friends': 86, 'age': 17}, ... {'name': 'Ted', 'friends': 23, 'age': 51} ... ] ... >>> import fastavro >>> parsed_schema1 = fastavro.parse_schema(schema1) >>> with open('example.file1.avro', 'wb') as f1: ... fastavro.writer(f1, parsed_schema1, records1) ... >>> # now demonstrate the use of fromavro() >>> import petl as etl >>> tbl1 = etl.fromavro('example.file1.avro') >>> tbl1 +-------+---------+-----+ | name | friends | age | +=======+=========+=====+ | 'Bob' | 42 | 33 | +-------+---------+-----+ | 'Jim' | 13 | 69 | +-------+---------+-----+ | 'Joe' | 86 | 17 | +-------+---------+-----+ | 'Ted' | 23 | 51 | +-------+---------+-----+ .. versionadded:: 1.4.0 )sourcelimitskipsN)rAvroView)rrr avro_argssource2rrE/var/www/Datamplify/venv/lib/python3.10/site-packages/petl/io/avro.pyfromavrosJr deflatec Ks"t|f|d||||d|dS)a Write the table into a new avro file according to schema passed. This method assume that each column has values with the same type for all rows of the source `table`. `Apache Avro`_ is a data serialization framework. It is used in data serialization (especially in Hadoop ecosystem), for dataexchange for databases (Redshift) and RPC protocols (like in Kafka). It has libraries to support many languages and generally is faster and safer than text formats like Json, XML or CSV. The `target` argument is the file path for creating the avro file. Note that if a file already exists at the given location, it will be overwritten. The `schema` argument (dict) defines the rows field structure of the file. Check fastavro `documentation`_ and Avro schema `reference`_ for details. The `sample` argument (int, optional) defines how many rows are inspected for discovering the field types and building a schema for the avro file when the `schema` argument is not passed. The `codec` argument (string, optional) sets the compression codec used to shrink data in the file. It can be 'null', 'deflate' (default), 'bzip2' or 'snappy', 'zstandard', 'lz4', 'xz' (if installed) The `compression_level` argument (int, optional) sets the level of compression to use with the specified codec (if the codec supports it) Additionally there are support for passing extra options in the argument `**avro_args` that are fowarded directly to fastavro. Check the fastavro `documentation`_ for reference. The avro file format preserves type information, i.e., reading and writing is round-trippable for tables with non-string data values. However the conversion from Python value types to avro fields is not perfect. Use the `schema` argument to define proper type to the conversion. The following avro types are supported by the schema: null, boolean, string, int, long, float, double, bytes, fixed, enum, :ref:`array `, map, union, record, and recursive types defined in :ref:`complex schemas `. Also :ref:`logical types ` are supported and translated to coresponding python types: long timestamp-millis, long timestamp-micros, int date, bytes decimal, fixed decimal, string uuid, int time-millis, long time-micros. Example usage for writing files:: >>> # set up a Avro file to demonstrate with >>> table2 = [['name', 'friends', 'age'], ... ['Bob', 42, 33], ... ['Jim', 13, 69], ... ['Joe', 86, 17], ... ['Ted', 23, 51]] ... >>> schema2 = { ... 'doc': 'Some people records.', ... 'name': 'People', ... 'namespace': 'test', ... 'type': 'record', ... 'fields': [ ... {'name': 'name', 'type': 'string'}, ... {'name': 'friends', 'type': 'int'}, ... {'name': 'age', 'type': 'int'}, ... ] ... } ... >>> # now demonstrate what writing with toavro() >>> import petl as etl >>> etl.toavro(table2, 'example.file2.avro', schema=schema2) ... >>> # this was what was saved above >>> tbl2 = etl.fromavro('example.file2.avro') >>> tbl2 +-------+---------+-----+ | name | friends | age | +=======+=========+=====+ | 'Bob' | 42 | 33 | +-------+---------+-----+ | 'Jim' | 13 | 69 | +-------+---------+-----+ | 'Joe' | 86 | 17 | +-------+---------+-----+ | 'Ted' | 23 | 51 | +-------+---------+-----+ .. versionadded:: 1.4.0 .. _Apache Avro: https://avro.apache.org/docs/current/spec.html .. _reference: https://avro.apache.org/docs/current/spec.html#schemas .. _documentation : https://fastavro.readthedocs.io/en/latest/writer.html wb)targetmodeschemasamplecodeccompression_levelN _write_toavro)tabler#r%r&r'r(rrrrtoavrocsa r,cKst|f|d||d|dS)aW Append rows into a avro existing avro file or create a new one. The `target` argument can be either an existing avro file or the file path for creating new one. The `schema` argument is checked against the schema of the existing file. So it must be the same schema as used by `toavro()` or the schema of the existing file. The `sample` argument (int, optional) defines how many rows are inspected for discovering the field types and building a schema for the avro file when the `schema` argument is not passed. Additionally there are support for passing extra options in the argument `**avro_args` that are fowarded directly to fastavro. Check the fastavro documentation for reference. See :meth:`petl.io.avro.toavro` method for more information and examples. .. versionadded:: 1.4.0 za+b)r#r$r%r&Nr))r+r#r%r&rrrr appendavros r-c@sHeZdZdZddZddZddZdd Zd d Zd d Z ddZ dS)rz;Read rows from avro file with their types and logical typescKs"||_||_||_||_d|_dSN)rrrr avro_schema)selfrrrrrrr__init__s  zAvroView.__init__cCs|jS)z*gets the schema stored in avro file header)r/)r0rrrget_avro_schemaszAvroView.get_avro_schemaccsf|jd"}||}||}|V|||D]}|VqWddS1s,wYdS)Nrb)ropen _open_reader_decode_schema_read_rows_from)r0 source_file avro_readerheaderrowrrr__iter__s  "zAvroView.__iter__cCs ddl}|j|fi|j}|S)z@This could raise a error when the file is corrupt or is not avrorN)fastavroreaderr)r0r8r=r9rrrr5szAvroView._open_readercCs6|j|_|jdur dS|jd}tdd|D}|S)z9extract the header from schema stored in avro file headerNNNfieldscss|]}|dVqdS)nameNr.0colrrr sz*AvroView._decode_schema..) writer_schemar/tuple)r0r9 schema_fieldsr:rrrr6s   zAvroView._decode_schemaccsfd}|jdur |jntj}t|D]\}}||jkrq||kr#dS|d7}|||}|VqdS)Nr)rsysmaxsize enumerater _map_row_from)r0r9r:countmaximumirecordr;rrrr7s  zAvroView._read_rows_fromcs6|dustrt}|Stfdd|D}|S)aV fastavro auto converts logical types defined in avro schema to correspoding python types. E.g: - avro type: long logicalType: timestamp-millis -> python datetime - avro type: int logicalType: date -> python date - avro type: bytes logicalType: decimal -> python Decimal Nc3s|]}|VqdSr.getrBrQrrrE0sz)AvroView._map_row_from..)r rGvalues)r0r:rQrrrTrrM$s  zAvroView._map_row_fromN) __name__ __module__ __qualname____doc__r1r2r<r5r6r7rMrrrrrs rc KsX|durdS|st||\}}nt||}trt|nt|} t||d} | |x} ddlm} ddl m } | |}| d| |||d|}d}| D]J}z | ||d}WqKt yw}zt |||||}tt |WYd}~qKd}~wty}zt |||||}tt|WYd}~qKd}~ww|WddS1swYdS)N)r$r) parse_schema)Writer)for%r'r(rIr)_build_schema_from_values_fix_missing_headersr r_ordered_dict_iteratorrr4r=r[fastavro.writer\write ValueError_get_error_details _raise_error TypeErrorflush)r+r#r$r%r&r'r(rtable2rowstarget2 target_filer[r\ parsed_schemawriternumrQverrvmsgterrtmsgrrrr*4sD        "r*cCs>t||d\}}t|}t|d}t||}t|}||fS)NrI)rrr _build_schema_fields_from_values_build_schema_with)r+r&samplesrhpropspeekrH schema_sourcerrrr^_s   r^cCsdddd|d}|S)NrQoutputavrozgenerated by petl)typerA namespacedocr@r)rHr%rrrrtisrtcCsDt}d}t}|D] }t|||||d}q dd|D}|S)NTFcSg|]}|qSrrrCitemrrr z4_build_schema_fields_from_values..)r_update_field_defs_fromrU)rwrvprevious fill_missingr@r;rHrrrrstsrsc Cst||D]a\}}|durdS||d}||d}d} t|tr1t|||||\} } } nt|||\} } | durG|d| gd||<n |rR|ddgd||<| dur\| ||d<| durf| ||d<qdS)N_prec_propnull)rAr{string)r rS isinstancedict_get_definition_from_record_get_definition_from_type_of) rvr;r@rrpropvaldprevfprevfcurrtdefdcurrrrrrs&   rcCs d}d}t|trddd}||fSt|tr ddd}||fSt|tr.ddd}||fSt|trGt|||\}}}dd||d }||fSt|trRd}||fSt|trct|||\}}||fSt|t rnd }||fSt|t ryd }||fSt|t rd}||fS|durd }||fSd S)Nlongztimestamp-millis)r{ logicalTypeintz time-millisrbytesdecimal)r{r precisionscalebooleandoublerr?) rrrrr _get_precision_from_decimalrlist_get_definition_from_arrayboolfloatr)rrprevrcurrrrrrrrsH          rc Csfd}t|D]}|dur qt|||\}}|dur|}|dur!|}q|dur(dn|}d|d}||fS)Nrarray)r{items)iterr) rrrafieldrfield2curr2bfieldrrrrrs  rc Csr|durt}|durt}t|}t|}t|||||dd|D}d|dd|d}|||fS)NcSr~rrrrrrrrz/_get_definition_from_record..rQ_recordrz)r{rAr|r@)rrkeysrUr) rrrrrrvr;rHrrrrrs   rcCst|dur d}}nt|\}}}}|dur+|d|d}}t||t||}}t|d}||d}|||fS)Nrrr)rr)precision_and_scalerSmax)rrrprecr_prec0scale0rrrrs    rc Cs|\}}}d}|D]}|d|}q d}d||}t|}|d}|dd} |r1| }tttt|} t|} | | | |fS)Nr rIr)as_tupler bit_lengthmathceillog10abs) numeric_valuesigndigitsexpnumberdigitdeltainumberbits_req bytes_reqrrrrrrs    rcCsT|dusd|vr |St|d\}}t|}t|}t|t|kr#|St||}|S)z'add missing columns headers from schemaNr@)rr_get_schema_header_nameslenr)r+r%r&rhcolsheaderstable3rrrr_s r_c Csht|}t|tr|t|g}n||g}t|}t|tr%d|nd}d} | |||||f} | S)z6show last row when failed writing for throubleshootingz output: %sz/failed writing on row #%d: %s %s schema: %s %s)rrrrrUrlookr ) r#rnerrrQr%rr+exampledestprinteddetailsrrrrd s  rdcCs(|d}|dur gSdd|D}|S)Nr@cSsg|]}|dqS)rArR)rCfieldrrrrsz,_get_schema_header_names..rR)r%r@r:rrrrs rcCsDt}|d}ztr|||tdWd}d}dSd}d}w)z8Works like raise Excetion(msg) from prev_exp in python3.rz%raise ErrorType, new_message, tracebkN)rJexc_infor with_tracebackexec) ErrorType new_messageexinftracebkrrrre#s rec cst|}t|}dd|D}|D],}t}t|D]\}}z||}Wn ty0d}Ynw|||fqt|VqdS)NcSsg|]}t|qSr)r )rCfrrrr4sz*_ordered_dict_iterator..)rnextrrL IndexErrorappendr) r+ithdrfldsr;rrPrvrrrr`1s   r`)Nr)Nr r!N)Nr )r!N)1 __future__rrrrJr collectionsrrrrrr petl.compatr r r r petl.io.sourcesrrpetl.transform.headersrrpetl.util.baserrrrrrr,r-rr*r^rtrsrrrrrrr_rdrrer`rrrrsD   Q k$C +