gVk,dZddlmZmZddlZddlmZddlmZm Z ddl Z ddl Z ddl Z ddl mZmZmZmZmZmZddlmZdd lmZdd lmZdd lmZdd lmZdd lmZm Z m!Z!m"Z"ddl#m$Z$ddl%m&Z&m'Z'm(Z(ddl)m*Z*ddl+m,Z,m-Z-dZ.dZ/dZ0dZ1dZ2dZ3dZ4dZ5dZ6dZ7dZ8dZ9dZ:dZ;d Zd2Z?d3Z@d4ZAd5ZBd6ZCe@eAeBeCd7ZDd8ZEd9ZF dd:lGmHZHeHjZJeFeDd;<d>ZN ddlOZOeNeDd?<d@ZP ddlQZRePeDdA<GdBdCeZSGdDdEeSZTGdFdGeSZUd!dHe zddIdJdfdIdIdIdKdLeeefdMe,dNeedOeVdPeWdQeeeVeVfdReXdSeYdTeeWdUeXdVeXdWeXfdXZZdIdIdIdKdLedMe,d/edUeXdVeXdWeXf dYZ[y#eK$r: ddlHZHeHjZJedwrite_item_count write_dataend_itemwrite_array_end)r"r#r$r%r&r'dtypeitems r( write_arrayrN`sl  5zA~  U,wD we]E7 K     r)c |jt|dkDrX|jt||d}|jD]&\}}|j |t ||||||(|j y)aMaps are encoded as a series of blocks. Each block consists of a long count value, followed by that many key/value pairs. A block with count zero indicates the end of the map. Each item is encoded per the map's value schema. If a block's count is negative, then the count is followed immediately by a long block size, indicating the number of bytes in the block. The actual count in this case is the absolute value of the count written.rvaluesN)write_map_startr>rHrFr9rI write_map_end) r"r#r$r%r&r'vtypekeyvals r( write_maprVtsy  5zA~  U,x  HC   s # wUM5' J& r)c d}t|trx|jdsg|\}}t|D])\}} t | } | t vr| d} n| } || k(s'|}n|dk(r"|rd|nd} d|d|d| z} t | |}nt|}d}d }t|D]\}} |rt | d k(r|}nt|| |d d| s-t | }||vr||} t | }|d k(rrt| }|r tj|}|r ||| }td | dD}t|}t|j|}||kDs|}|}|dk(r|}d}|}n|dk(r(|rd|nd} t t|d|d|d| |}|j|||t!|||||||y)zA union is encoded by first writing a long value indicating the zero-based position within the union of the schema of its value. The value is then encoded per the indicated schema within the union.disable_tuple_notationnamez on field zprovided union type name z not found in schema  Fdouble) raise_errorsfieldr'recordc3&K|] }|d ywrZN).0fs r( zwrite_union..s*R>Q1V9>QfieldsfloatTz (type z) do not match N) isinstancetupleget enumeraterrr?typerrrsetr> intersectionrepr write_indexrI)r"r#r$r%r&r'best_match_indexrZrC candidateextracted_type schema_namer_msgpytype most_fieldscould_be_float record_type logical_typepreparecandidate_fields datum_fieldsrhs r( write_unionrs\ % 4L(M u )& 1 E90;N,'/ , {"#( !2 r !+0iw'bE+D61FGHAeW%& S/ ! e  )& 1 E9&y1X=',$" 2)< -/ -k :I"5i"@K(*#7 #BL#"1"5"5l"C"$+E9$=E'**Ri>Q*R'R$#&u:L !1!>!>|!LMF ++0(&,  G+',$&*N',$W!2X r !+0iw'bE;-wvhofXQugN ! ve}-wve}mUGLr)c t|td|dDz }|jds|jdr|rtddj||dD]}|d}|d} ||vrK|jds|jdrd |vrtd |d d |vrd | vrtd ||j||jd } | dk(s| dk(r t | } t || | |||y)zA record is encoded by encoding the values of its fields in the order that they are declared. In other words, a record is encoded as just the concatenation of the encodings of its fields. Field values are encoded per their schema.c3&K|] }|d ywrbrc)rdr_s r(rfzwrite_record..sJ9IeFm9Irgrhstrictstrict_allow_defaultz7record contains more fields than the schema specifies: z, rZrndefaultzField z7 is specified in the schema but missing from the recordnullzno value and no default for rir]N)rorlr?joinrirI) r"r#r$r%r&r'extrasr_rZ field_type datum_values r( write_recordrs3 Z#J9IJJ JF H-C!D&EdiiPVFWEX Y  !V}6] u {{8$ 23 8N TF"YZ%'F*,D #?v!FGGiieii &:;  J($: ,K        !"r)rbooleanstringintlongrir]bytesfixedenumarraymapunion error_unionr`errorc*t|}t|}tj|}|r0|r t j|} | r | ||} |||||||St |||||d|S#t $r} |rt | d|d} ~ wwxYw)a&Write a datum of data to output stream. Parameters ---------- encoder: encoder Type of encoder (e.g. binary or json) datum: object Data to write schema: dict Schema to use named_schemas: dict Mapping of fullname to schema definition z on field Nr[)rrWRITERSrlr TypeErrorrI) r"r#r$r%r&r'r{r|fnr}exs r(rIrI#s&f-K'/L [ !B %)),7Gv. gufmUGL L  UM+6 r7   2$j 899  s A11 B:B  Bct|jDcic]\}}||jc}}|d}t||tidiycc}}w)N)magicmetasyncr[)rrFencoderIr)r"metadata sync_markerrTvalueheaders r( write_headerrGsS7?~~7GH7Geelln$7GHF w r2r:IsAcn|jt||jj|y)zWrite block in "null" codec.N)r0r>_fowrite)r" block_bytescompression_levels r(null_write_blockrPs' s;'( KKk"r)c|tj||dd}ntj|dd}|jt||jj |y)zWrite block in "deflate" codec.NrX)zlibcompressr0r>rrr"rrdatas r(deflate_write_blockrVs[$}}[*;rrrs r(bzip2_write_blockrbs4 << $D s4y! KKdr)ctj|}|jt||jj |y)zWrite block in "xz" codec.N)lzmarr0r>rrrs r(xz_write_blockris4 == %D s4y! KKdr))rdeflatebzip2xzcfd}|S)Nc,tddz)Nz7 codec is supported but you need to install one of the zfollowing libraries: )r?)r"rrcodec librariess r(missingz#_missing_codec_lib..missingys*gL M%i[1 2  r)rc)rrrs`` r(_missing_codec_librxs Nr)ct|}|jt|dz|jj ||j |y)zWrite block in "snappy" codec.N)snappy_compressr0r>rrr;rs r(snappy_write_blockrsB ; 'D s4y1}% KKd  $r))snappyrzbSnappy compression will use `cramjam` in the future. Please make sure you have `cramjam` installedcramjamc|&tj|j|}n#tjj|}|jt ||j j |y)z!Write block in "zstandard" codec.N)level) zstandardZstdCompressorrr0r>rrrs r(zstandard_write_blockrsa$''.?@II+V'')22;? s4y! KKdr)rctjj|}|jt ||j j |y)zWrite block in "lz4" codec.N)lz4blockrr0r>rrrs r(lz4_write_blockrs: 99  k *D s4y! KKdr)rc:eZdZddifdZedZedZy) GenericWriterNc Bi|_|rtnd|_|xsi|_||_|t ||j|_t|tr(|jDcic] \}}|dvr||}}}ntt|trdg}|D][}t|tr8|j|jDcic] \}}|dvr||c}}K|j|]|}tj||jd<ycc}}wcc}}w)N)__fastavro_parsed__named_schemasz avro.schema)_named_schemasr validate_fnrr'rr$rjdictrFlistappendjsondumps) selfr$r validatorr'rTrschemasss r(__init__zGenericWriter.__init__s& (19t B    &vt/B/BCDK fd ##),,."0JCFFU "0   %Ga&NN/0ggi.7 U"$  J.7 NN1%F'+zz&'9 m$1s )DD cyNrcrr`s r(rzGenericWriter.write r)cyrrcrs r(flushzGenericWriter.flushrr))__name__ __module__ __qualname__rrrrrcr)r(rrs7(,b$:L    r)rceZdZddezddddifdeeefdeded e d e e eefd e d e d e e de ee fffd ZdZdZdZdZxZS)WriterrNFr)for$r sync_intervalrrrrr'c t ||||| ||jd<t|tr||_nt ||_t t |_d|_||_ ||_ t|j jr|j jjdt|j j} | j} i|_t#| j$|j |_| jj)dd}| d|_|j jjddt,||_y|xst1t2|_ t,||_t9|j |j|j*y#t4$rt7d|wxYw)Nz avro.codecrrrrzunrecognized codec: )superrrrjrr"rio block_countrrrrseekr_headerrr writer_schemar$rlr BLOCK_WRITERS block_writerrrKeyErrorr?r) rrr$rrrrrrr' avro_readerr __class__s r(rzWriter.__init__s 9g>&+ l# b- (DL(,DL **!2 $,,** + LL   ! !! $ !1!12K ((F"$D &{'@'@$BUBUVDK((,,\6BE%f~D  LL   ! !!Q ' -e 4D *@gi.@D  A$1%$8! t}}d6F6F G A #7w!?@@ As GGc|jj|j|j|j|jj j |j|jj j|j|jj jd|jj jdtd|_yNr) r"r0rrrrgetvaluerrrtruncaterrrs r(dumpz Writer.dump$s  0 01 $,, (<(<(>@V@VW t//0  Q  H%r)c|jr4|j||j|jdd|jt |j ||j|jd|j|xj dz c_|j jj|jk\r|jyy)Nr[Tr) rr$rr'rIrrrtellrrrs r(rz Writer.write,s       T%8%8"dDLL  477FDKK1D1Db$,,W A 77;;   !3!3 3 IIK 4r)c|jjjs|jdkDr|j |j j |j|j|j |jj|j|j jj|jyr)rrrrrr"r0 num_recordsrbytes_rrrr)rrs r( write_blockzWriter.write_block6s 77;;   !1!1A!5 IIK  1 12 $,, (=(=(?AWAWX t//0r)c|jjjs|jdkDr|j |j jj yr)rrrrrr"rrs r(rz Writer.flush>sB 77;;   !1!1A!5 IIK  r))rrrrrr rrstrrr r boolrrrrrr __classcell__rs@r(rrs !I--1 +/#%1H "m# $1H1H 1H  1H 4S>* 1H1H1H$C=1Hc4i1Hf1!r)rceZdZddezddddifdededed ed ee eefd e d e d eede ee fffd Z dZ dZxZS) JSONWriterrrNFr)rr$rrrrrcodec_compression_levelr'c t ||||| ||_|jj|j|j yr)rrr" configurer$r) rrr$rrrrrrr'rs r(rzJSONWriter.__init__Es= 9g>  t{{D,?,?@r)c|jr4|j||j|jdd|jt |j ||j|jd|jy)Nr[T)rr$rr'rIr"rs r(rzJSONWriter.writeVs]       T%8%8"dDLL   LL&$++t/B/BB  r)c8|jjyr)r"rrs r(rzJSONWriter.flush_s r))rrrrrrr rr r r rrrrr r s@r(rrDs !I--1 15#%A AA A  A 4S>* AAA"*#Ac4iA" r)rrFr)rrrYrr$recordsrrrrrrrrrYc t|tr tdt|trt ||||||||| | | d } n!t t ||||||||| | | d } |D]} | j| | jy)av Write records to fo (stream) according to schema Parameters ---------- fo Output stream schema Writer schema records Records to write. This is commonly a list of the dictionary representation of the records, but it can be any iterable codec Compression codec, can be 'null', 'deflate' or 'snappy' (if installed) sync_interval Size of sync interval metadata Header metadata validator If true, validation will be done on the records sync_marker A byte string used as the avro sync marker. If not provided, a random byte string will be used. codec_compression_level Compression level to use with the specified codec (if the codec supports it) strict If set to True, an error will be raised if records do not contain exactly the same fields that the schema states strict_allow_default If set to True, an error will be raised if records do not contain exactly the same fields that the schema states unless it is a missing field that has a default value in the schema disable_tuple_notation If set to True, tuples will not be treated as a special case. Therefore, using a tuple to indicate the type of a record will not work Example:: from fastavro import writer, parse_schema schema = { 'doc': 'A weather reading.', 'name': 'Weather', 'namespace': 'test', 'type': 'record', 'fields': [ {'name': 'station', 'type': 'string'}, {'name': 'time', 'type': 'long'}, {'name': 'temp', 'type': 'int'}, ], } parsed_schema = parse_schema(schema) records = [ {u'station': u'011990-99999', u'temp': 0, u'time': 1433269388}, {u'station': u'011990-99999', u'temp': 22, u'time': 1433270389}, {u'station': u'011990-99999', u'temp': -11, u'time': 1433273379}, {u'station': u'012650-99999', u'temp': 111, u'time': 1433275478}, ] with open('weather.avro', 'wb') as out: writer(out, parsed_schema, records) The `fo` argument is a file-like object so another common example usage would use an `io.BytesIO` object like so:: from io import BytesIO from fastavro import writer fo = BytesIO() writer(fo, schema, records) Given an existing avro file, it's possible to append to it by re-opening the file in `a+b` mode. If the file is only opened in `ab` mode, we aren't able to read some of the existing header information and an error will be raised. For example:: # Write initial records with open('weather.avro', 'wb') as out: writer(out, parsed_schema, records) # Write some more records with open('weather.avro', 'a+b') as out: writer(out, None, more_records) Note: When appending, any schema provided will be ignored since the schema in the avro file will be re-used. Therefore it is convenient to just use None as the schema. z2"records" argument should be an iterable, not dictr)r'N) rjrr?rrrrrr)rr$rrrrrrrrrrYoutputr`s r(writerrcsV'4 MNN"o&        # (<*@   "        # (<*@   V LLNr)c ~i}t||}t|}t||||d|||d|jy)aWrite a single record without the schema or header information Parameters ---------- fo Output file schema Schema record Record to write strict If set to True, an error will be raised if records do not contain exactly the same fields that the schema states strict_allow_default If set to True, an error will be raised if records do not contain exactly the same fields that the schema states unless it is a missing field that has a default value in the schema disable_tuple_notation If set to True, tuples will not be treated as a special case. Therefore, using a tuple to indicate the type of a record will not work Example:: parsed_schema = fastavro.parse_schema(schema) with open('file', 'wb') as fp: fastavro.schemaless_writer(fp, parsed_schema, record) Note: The ``schemaless_writer`` can only write a single record. r[rN)rrrIr)rr$r`rrrYr%r"s r(schemaless_writerrsPN#%M &- 0FBG $8&<  MMOr))\__doc__abcrrrrrosrrrrrtypingrr r r r r warningsrconstrio.binary_encoderrio.json_encoderr validationrreadrrrrlogical_writersrr$rrr _write_commonrtypesrrr r+r.r0r2r4r6r9r;r@rDrNrVrrrrIrrrrrrrrrr compress_rawr ImportErrorrDeprecationWarningrrr lz4.blockrrrrr rr rrrrcr)r(r-s( $  ;;,,!99,KK)' !       ((`MF! H  J  }  j  9    J    [   l  [  [  J  [  9  [ ;  l   \! (! H;#  "    %1))O1M(7"7M+++M%- C- `Q!]Q!hF ))--1R!&#(R b/!"R Rc]R  R  R tCH~& RRR&c]R RR!Rt!&#(7 7 7 7  7  7!7U  5 5 // p  #5 h J"4Xy"I hJ 5:N!3K!MM+N<-eU;M%