Tuesday, November 14, 2023

Server Send Events Stream SSE + Javascript

from quart import abort, make_response, request, Quart

app = Quart(__name__)
@app.get("/sse")
async def sse():
    if "text/event-stream" not in request.accept_mimetypes:
        abort(400)
    response = await make_response(
        send_events(),
        {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Transfer-Encoding': 'chunked',
        },
    )
    response.timeout = None  # Disable the timeout for this streaming response
    return response
 
<script>
        document.addEventListener('DOMContentLoaded', (event) => {
            const eventSource = new EventSource('/sse');
            eventSource.onmessage = function(event) {
                console.log('New message:', event.data);
                // You can update the DOM here
            };
            eventSource.onopen = function(event) {
                console.log('Connection to server opened.');
            };
            eventSource.onerror = function(event) {
                console.error('EventSource failed.');
            };
            // To close the connection when the window is closed
            window.onbeforeunload = () => {
                eventSource.close();
            };
        });
    </script>
 
With parameters:
from quart import Quart, request, abort, make_response
from dataclasses import dataclass

app = Quart(__name__)

@dataclass
class ServerSentEvent:
data: str
event: str | None = None
id: int | None = None
retry: int | None = None

def encode(self) -> bytes:
message = f"data: {self.data}"
if self.event is not None:
message = f"{message}\nevent: {self.event}"
if self.id is not None:
message = f"{message}\nid: {self.id}"
if self.retry is not None:
message = f"{message}\nretry: {self.retry}"
message = f"{message}\n\n"
return message.encode('utf-8')

@app.get("/chat-updates")
async def chat_updates():
if "text/event-stream" not in request.accept_mimetypes:
abort(400)

friends = request.args.get('friends', default='false') == 'true'
format = request.args.get('format', default='simple')

async def send_events():
while True:
# Your logic to get updates, possibly filtering based on the query parameters
data = ... # Replace with your data retrieval logic
event = ServerSentEvent(data)
yield event.encode()

response = await make_response(
send_events(),
{
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Transfer-Encoding': 'chunked',
},
)
response.timeout = None
return response

# Run the app
if __name__ == "__main__":
app.run()
Javascript:
<script>
const params = new URLSearchParams({
friends: 'true',
format: 'detailed'
});

const url = `https://my-server/chat-updates?${params}`;
const eventSource = new EventSource(url);

eventSource.onmessage = function(event) {
// Handle incoming messages
console.log(event.data);
};

eventSource.onerror = function(error) {
// Handle any errors that occur
console.error("EventSource failed:", error);
}; 
</script>
 
// server.js
const http = require('http');
const es = require('event-stream');
// Create a HTTP server
const server = http.createServer((req, res) => {
// Check if the request path is /stream
if (req.url === '/stream') {
// Set the response headers
res.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
'Connection': 'keep-alive'
});
// Create a counter variable
let counter = 0;
// Create an interval function that sends an event every second
const interval = setInterval(() => {
// Increment the counter
counter++;
// Create an event object with name, data, and id properties
const event = {
name: 'message',
data: `Hello, this is message number ${counter}`,
id: counter
};
// Convert the event object to a string
const eventString = `event: ${event.name}\ndata: ${event.data}\nid: ${event.id}\n\n`;
// Write the event string to the response stream
res.write(eventString);
// End the response stream after 10 events
if (counter === 10) {
clearInterval(interval);
res.end();
}
}, 1000);
} else {
// Handle other requests
res.writeHead(404);
res.end('Not found');
}
});
// Listen on port 3000
server.listen(3000, () => {
console.log('Server listening on port 3000');
});
 
// client.js
// Fetch the event stream from the server
fetch('/stream')
.then(response => {
// Get the readable stream from the response body
const stream = response.body;
// Get the reader from the stream
const reader = stream.getReader();
// Define a function to read each chunk
const readChunk = () => {
// Read a chunk from the reader
reader.read()
.then(({
value,
done
}
) =>
{
// Check if the stream is done
if (done) {
// Log a message
console.log('Stream finished');
// Return from the function
return;
}
// Convert the chunk value to a string
const chunkString = new TextDecoder().decode(value);
// Log the chunk string
console.log(chunkString);
// Read the next chunk
readChunk();
})
.catch(error => {
// Log the error
console.error(error);
});
};
// Start reading the first chunk
readChunk();
})
.catch(error => {
// Log the error
console.error(error);
});
  
 
from flask import Flask, Response, render_template
import itertools
import time

app = Flask(__name__)

@app.route('/')
def index():
return render_template('index.html')

@app.route('/connect')
def publish_hello():
def stream():
for idx in itertools.count():
msg = f"data: <p>This is {idx}.</p>\n\n"
yield msg
time.sleep(1)
return Response(stream(), mimetype="text/event-stream")

Htmx

<!DOCTYPE html>
<html>
<head>
<script src="https://unpkg.com/htmx.org@1.8.6"></script>
<script src="https://unpkg.com/htmx.org/dist/ext/sse.js"></script>
</head>
<body>
<div hx-ext="sse" sse-connect="/connect" sse-swap="message">
Contents of this box will be updated in real time with every SSE message received from the server.
</div>
</body>
</html>
 
In this example, the Flask app sends a message to the HTMX client every second. The message is wrapped in an HTML paragraph tag and sent as an SSE event. The HTMX client listens to the SSE endpoint and updates the contents of the HTML element with the sse-swap attribute with the message received from the server.
Streaming data from Flask to HTMX using Server-Side Events (SSE) | mathspp 
 // Client-side Javascript in the HTML 
var targetContainer = document.getElementById("this-div");
var eventSource = new EventSource("/stream");
eventSource.onmessage = function(e) {
targetContainer.innerHTML = e.data;
};

Generator + SSE: So why are Python generators good with SSE? It’s simply because they can keeping looping and yielding data and handing it to the client very seamlessly. Here is a simple Python implementation of SSE in Flask:

@route("/stream")
def stream():
def eventStream():
while True:
# Poll data from the database
# and see if there's a new message
if len(messages) > len(previous_messages):
yield "data:
{}\n\n".format(messages[len(messages)-1)])"

return Response(eventStream(), mimetype="text/event-stream")

This is a simple hypothetical event source that checks if there’s a new inbox message and yield the new message. For the browser to acknowledge a server-sent message, you’ll have to comply to this format:

"data: <any_data>\n\n"

You have the option to also send with the data the event and id.

"id: <any_id>\nevent: <any_message>\ndata: <any_data>\n\n"

Note that the fields do not have to be in any order as long as there is a newline (\n) for each field and two (\n\n) at the end of them. With additional event field, you can have more control how you push data to the browser.

// Client-side Javascript in the HTMLvar targetContainer = document.getElementById("this-div");
var eventSource = new EventSource("/stream");
eventSource.addEventListener = (<any_message>, function(e) {
targetContainer.innerHTML = e.data;

if (e.data > 20) {
targetContainer.style.color = "red";
}
};

This will basically render the DOM with the latest data on the specified event message and change the color to “red” when it exceeds 20.

https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#receiving_events_from_the_server

Warning: When not used over HTTP/2, SSE suffers from a limitation to the maximum number of open connections, which can be especially painful when opening multiple tabs, as the limit is per browser and is set to a very low number (6).
The issue has been marked as "Won't fix" in Chrome and Firefox.
This limit is per browser + domain, which means that you can open 6 SSE connections
across all of the tabs to www.example1.com and another 6 SSE connections to www.example2.com (per Stackoverflow). When using HTTP/2, the maximum number of simultaneous HTTP streams is negotiated between the server and the client (defaults to 100). 

Fields
Each message received has some combination of the following fields, one per line:

event
    A string identifying the type of event described. If this is specified, an event will be dispatched on the browser to the listener for the specified event name; the website source code should use addEventListener() to listen for named events. The onmessage handler is called if no event name is specified for a message.

data
    The data field for the message. When the EventSource receives multiple consecutive lines that begin with data:, it concatenates them, inserting a newline character between each one. Trailing newlines are removed.

id
    The event ID to set the EventSource object's last event ID value.

retry
    The reconnection time. If the connection to the server is lost, the browser will wait for the specified time before attempting to reconnect. This must be an integer, specifying the reconnection time in milliseconds. If a non-integer value is specified, the field is ignored.

All other field names are ignored.

Sunday, November 5, 2023

Select and Delete Duplicates from Sqlite Table

Select duplicates from table PonsApimetaDE with columns DE, meta

SELECT DE, meta, COUNT(*)
FROM PonsApimetaDE
GROUP BY DE, meta
HAVING COUNT(*) > 1

Delete duplicates from table PonsApimetaDE with columns DE, meta

DELETE FROM PonsApimetaDE
WHERE rowid NOT IN (
  SELECT MIN(rowid)
  FROM PonsApimetaDE
  GROUP BY DE, meta
)

Delete duplicates from table

CREATE TABLE temp_table as SELECT DISTINCT * FROM source_table;
DELETE FROM source_table;
INSERT INTO source_table SELECT * FROM temp_table
DROP TABLE temp_table

Create index for table PonsApimetaDE with columns DE, meta

CREATE UNIQUE INDEX "idxuniquemetaDE" ON "PonsApimetaDE" (
  "DE"  ASC,
  "meta"  ASC
)

*****

SELECT * FROM table
WHERE rowid > (
  SELECT MIN(rowid) FROM table p2
  WHERE table.column1 = p2.column1
  AND table.column2 = p2.column2
);

DELETE FROM table
WHERE rowid > (
  SELECT MIN(rowid) FROM table p2
  WHERE table.column1 = p2.column1
  AND table.column2 = p2.column2
);

p2 is an alias for the table table in the subquery. This alias is used to make it easier to reference the table table in the subquery without having to type out the entire table name each time.

The purpose of this DELETE statement is to delete all rows from the table table where the column1 and column2 are the same as other rows, but only if the current row's rowid is greater than the smallest rowid with the same column1 and column2.

Here's a breakdown of the query:

The DELETE FROM table statement tells SQLite to delete all rows from the table table that meet the conditions specified in the WHERE clause.

The WHERE rowid > (subquery) clause restricts the rows to be deleted based on their rowid.

The subquery (SELECT MIN(rowid) FROM table p2 WHERE table.column1 = p2.column1 AND table.column2 = p2.column2) calculates the smallest rowid in the table table that has the same column1 and column2 as the current row.

The p2 alias in the subquery allows us to reference the table table without having to type out the entire table name each time.

In summary, this query will delete all duplicate rows from the table table, leaving only one row per unique combination of column1 and column2.

Saturday, November 4, 2023

Mongoose Custom Validator String or Number

const mongoose = require('mongoose');
const { Schema } = mongoose;

main().catch(err => console.log(err));

async function main() {
  await mongoose.connect('mongodb://127.0.0.1:27017/test');
  // use `await mongoose.connect('mongodb://user:password@127.0.0.1:27017/test');`
//   if your database has auth enabled
}
 
const stringornumbervalidator = function(value) {
  return typeof value === 'string' || typeof value === 'number';
};

const MySchema = new Schema({
  myField: {
    type: Schema.Types.Mixed,
    validate: {
      validator: stringornumbervalidator,
      message: 'Value should be either a string or a number'
    }
  }
});

const MyModel = mongoose.model('MyModel', MySchema);

Python Asynchronous Context Manager PlayGround

class AsyncContextManager:
    global correlationMap
    global requestId
    async def __aenter__(self) -> Literal['Waited']: # enter the async context manager
        print('>Entering the async context manager') # report a message
        await asyncio.sleep(0.05) # block for a moment
        return 'Waited'
    async def __aexit__(self, exc_type, exc, tb) -> Literal['Exited']: # exit the async context manager
        print('>Exiting the async context manager') # send a message
        await asyncio.sleep(0.05) # block for a moment
        return 'Exited'
    async def getresult(self):
        await asyncio.sleep(0.05)
        # print('Running inside the context manager')
        return correlationMap[requestId]
    async def dummyasync() -> Literal['Dummy function']:
        print(f"Dummy function started")
        await asyncio.sleep(0.1)
        return 'Dummy function' 
 
async with AsyncContextManager():
    app.add_background_task(callback)
    app.add_background_task(checkResult)
    while responsefirstcall.text == 'Waiting...':
        correlationMap[requestId] = await AsyncContextManager().getresult()
        print('correlationMap before loop: ' + str(correlationMap[requestId]))
            while correlationMap[requestId] == 'Waiting...':
                await asyncio.sleep(0.2)
                print('checkResult inside while loop: ' + str(response.text[:-1]))
                correlationMap[requestId] = await AsyncContextManager().getresult()
                print('Check result inside while loop: ' + str(correlationMap[requestId][:-1]) + ' ' + str(time.strftime("%A %d:%m:%Y %H:%M:%S UTC%z", time.localtime())))
try:
                            # response: httpx.Response = await client.get(f'http://{externalip}:5000/checkResult')
                            if correlationMap[requestId] != 'Waiting...':
                            # if response.text != 'Waiting...': # print('Not Waiting...')
                                # etransdomain = scripts.etrans.etrans(sourcelang, [targetlang], 'SPD', searchformterm).domain
                                # rows.append('E-Translate domain: ' + etransdomain)
                                rows.append('E-Translate')
                                rows.append(correlationMap[requestId].replace('\n', ''))
                                # rows.append(response.text.replace('\n', ''))
                                print(rows)
                                break
                            # else:
                            #     continue
                        except Exception as exc:
                            print(exc)
                            await asyncio.sleep(0.1)
                            rows.append('E-Translate')
                            rows.append('Error')
                            break
 
if rows: print("list is not empty")

Future

Future is simply an abstraction of value that may be not computed yet and will be available eventually. It's a simple container that only does one thing - whenever the value is set, fire all registered callbacks.

If you want to obtain that value, you register a callback via add_done_callback() method.

But unlike in Promise, the actual computation is done externally - and that external code has to call set_result() method to resolve the future.

Coroutine

Coroutine is the object very similar to Generator.

A generator is typically iterated within for loop. It yields values and, starting from PEP342 acceptance, it receives values.

A coroutine is typically iterated within the event loop in depths of asyncio library. A coroutine yields Future instances. When you are iterating over a coroutine and it yields a future, you shall wait until this future is resolved. After that you shall send the value of future into the coroutine, then you receive another future, and so on.

An await expression is practically identical to yield from expression, so by awaiting other coroutine, you stop until that coroutine has all its futures resolved, and get coroutine's return value. The Future is one-tick iterable and its iterator returns actual Future - that roughly means that await future equals yield from future equals yield future.

Task

Task is Future which has been actually started to compute and is attached to event loop. So it's special kind of Future (class Task is derived from class Future), which is associated with some event loop, and it has some coroutine, which serves as Task executor.

Task is usually created by event loop object: you give a coroutine to the loop, it creates Task object and starts to iterate over that coroutine in manner described above. Once the coroutine is finished, Task's Future is resolved by coroutine's return value.

You see, the task is quite similar to JS Promise - it encapsulates background job and its result.

Coroutine Function and Async Function

Coroutine func is a factory of coroutines, like generator function to generators. Notice the difference between Python's coroutine function and Javascript's async function - JS async function, when called, creates a Promise and its internal generator immediately starts being iterated, while Python's coroutine does nothing, until Task is created upon it.

API Calls In NodeJS Express

Create a new route in express called /getAPIResponse. This route will make a REST API call and return the response as JSON.

app.get('/getAPIResponse', (req, res) => {
    // API code will be here
})

Install request using npm: npm install --save request

Once you have the request module installed, create a file called API_helper.js. This will be wrapper for request module to make API calls. In future if you need to use any other module, you simple need to modify the API_helper.js wrapper and not every where inside the application.

Require the request module inside the API_helper.js. We’ll be returning a promise from the helper method, so here is how the API_helper.js looks :

const request = require('request')

module.exports = {
    /*
    ** This method returns a promise
    ** which gets resolved or rejected based
    ** on the result from the API
    */
    make_API_call : function(url){
        return new Promise((resolve, reject) => {
            request(url, { json: true }, (err, res, body) => {
              if (err) reject(err)
              resolve(body)
            });
        })
    }
}

As seen in the above code, the make_API_call method returns a promise which gets resolved or rejected based on API response.

Require the above module in the app.js file.

// require API_helper.js
const api_helper = require('./API_helper')

Make an API call to the REST API https://jsonplaceholder.typicode.com/todos/1 using api_helper which returns some dummy JSON data.

Here is the app.js file:

const express = require('express')
const api_helper = require('./API_helper')
const app = express()
const port = 3000

/*
* Route to DEMO the API call to a REST API Endpoint 
* REST URL : https://jsonplaceholder.typicode.com/todos/1
*/
app.get('/getAPIResponse', (req, res) => {
    api_helper.make_API_call('https://jsonplaceholder.typicode.com/todos/1')
    .then(response => {
        res.json(response)
    })
    .catch(error => {
        res.send(error)
    })
})

app.listen(port, () => console.log(`App listening on port ${port}!`))

api_helper returns a promise which when resolved returns a JSON or returns an error when rejected.

Save the above changes and restart the server. Point your browser to http://localhost:3000/getAPIResponse and will have the API response returned in the browser.

Wednesday, November 1, 2023

Truthy and Falsy Values in Python

In Python, individual values can evaluate to either True or False. They do not necessarily have to be part of a larger expression to evaluate to a truth value because they already have one that has been determined by the rules of the Python language.

The basic rules are:

  • Values that evaluate to False are considered Falsy.
  • Values that evaluate to True are considered Truthy.

According to the Python Documentation:

Any object can be tested for truth value, for use in an if or while condition or as operand of the Boolean operations below (and, or, not).

Boolean Context

When we use a value as part of a larger expression, or as an if or while condition, we are using it in a boolean context.

You can think of a boolean context as a particular "part" of your code that requires a value to be either True or False to make sense. 

Falsy Values

Sequences and Collections:

  • Empty lists []
  • Empty tuples ()
  • Empty dictionaries {}
  • Empty sets set()
  • Empty strings ""
  • Empty ranges range(0)

Numbers

  • Zero of any numeric type.
  • Integer: 0
  • Float: 0.0
  • Complex: 0j

Constants

  • None
  • False

Falsy values were the reason why there was no output in our initial example when the value of a was zero.

Truthy Values

According to the Python Documentation:

By default, an object is considered true.

Truthy values include:

  • Non-empty sequences or collections (lists, tuples, strings, dictionaries, sets).
  • Numeric values that are not zero.
  • True

The Built-in bool() Function

You can check if a value is either truthy or falsy with the built-in bool() function.

According to the Python Documentation, this function:

Returns a Boolean value, i.e. one of True or False. x (the argument) is converted using the standard truth testing procedure.
>>> bool(5)
True
>>> bool(0)
False
>>> bool([])
False
>>> bool({5, 5})
True
>>> bool(-5)
True
>>> bool(0.0)
False
>>> bool(None)
False
>>> bool(1)
True
>>> bool(range(0))
False
>>> bool(set())
False
>>> bool({5, 6, 2, 5})
True

 

__bool __()

With the special method __bool__(), you can set a "customized" condition that will determine when an object of your class will evaluate to True or False.

According to the Python Documentation:

By default, an object is considered true unless its class defines either a __bool__() method that returns False or a __len__() method that returns zero, when called with the object.

For example, if we have this very simple class:

>>> class Account:
	
	def __init__(self, balance):
		self.balance = balance

You can see that no special methods are defined, so all the objects that you create from this class will always evaluate to True:

>>> account1 = Account(500)
>>> bool(account1)
True
>>> account2 = Account(0)
>>> bool(account2)
True

We can customize this behavior by adding the special method __bool__():

>>> class Account:
	def __init__(self, balance):
		self.balance = balance
		
	def __bool__(self):
		return self.balance > 0

Now, if the account balance is greater than zero, the object will evaluate to True. Otherwise, if the account balance is zero, the object will evaluate to False.

>>> account1 = Account(500)
>>> bool(account1)
True
>>> account2 = Account(0)
>>> bool(account2)
False

💡 Tip: If __bool__() is not defined in the class but the __len__() method is, the value returned by this method will determine if the object is truthy or falsy.

🔹 In Summary

  • Truthy values are values that evaluate to True in a boolean context.
  • Falsy values are values that evaluate to False in a boolean context.
  • Falsy values include empty sequences (lists, tuples, strings, dictionaries, sets), zero in every numeric type, None, and False.
  • Truthy values include non-empty sequences, numbers (except 0 in every numeric type), and basically every value that is not falsy.
  • They can be used to make your code more concise. 

Source: freecodecamp.org

 

Sunday, October 29, 2023

Word Macro Insert Page X of Y

Sub InsertTotalPages()
' Inserts total number of pages in the document at cursor
    Selection.Fields.Add Range:=Selection.Range, Type:=wdFieldEmpty, Text:= _
        "NUMPAGES  ", PreserveFormatting:=True
End Sub
 

Sub InsertPageXofY()
' Inserts Page {current_page} of {total_doc_pages} at cursor
    Selection.TypeText Text:="Page "
    Selection.Fields.Add Range:=Selection.Range, Type:=wdFieldPage, Text:="Page ", _
        PreserveFormatting:=True
    Selection.TypeText Text:=" of "
    Selection.Fields.Add Range:=Selection.Range, Type:=wdFieldEmpty, Text:= _
        "NUMPAGES  ", PreserveFormatting:=True
End Sub

Friday, October 27, 2023

Python PlatitorTvaRest API

from datetime import datetime
import requests, json

headers = {
    'Accept': 'application/json',
    'Accept-Language': 'en-US,en;q=0.7,ro;q=0.3',
    'Connection': 'keep-alive',
    'Referer': 'https://mfinante.gov.ro',
    'Content-Type': 'application/json',
}
today = datetime.today().strftime('%Y-%m-%d')
data = [{'cui': '19467555', 'data': today}]
print(data)

response = requests.post('https://webservicesp.anaf.ro/PlatitorTvaRest/api/v8/ws/tva', data=json.dumps(data), headers=headers)
print(response.json())

 *****

[{'cui': '19467555', 'data': '2023-10-27'}]

{
  "cod": 200,
  "message": "SUCCESS",
  "found": [
    {
      "date_generale": {
        "cui": 19467555,
        "data": "2023-10-27",
        "denumire": "LEON TIBERIU CRISTIAN PERSOANĂ FIZICĂ AUTORIZATĂ",
        "adresa": "JUD. BRAŞOV, MUN. SĂCELE, STR. VALEA CERNATULUI, NR.47",
        "nrRegCom": "F08/388/2003",
        "telefon": "0729947925",
        "fax": "",
        "codPostal": "505600",
        "act": "",
        "stare_inregistrare": "RELUARE ACTIVITATE din data 30.04.2013",
        "data_inregistrare": "2007-01-01",
        "cod_CAEN": "7430",
        "iban": "",
        "statusRO_e_Factura": false,
        "organFiscalCompetent": "Administraţia Judeţeană a Finanţelor Publice Braşov",
        "forma_de_proprietate": "",
        "forma_organizare": "",
        "forma_juridica": ""
      },
      "inregistrare_scop_Tva": {
        "scpTVA": false,
        "perioade_TVA": []
      },
      "inregistrare_RTVAI": {
        "dataInceputTvaInc": "",
        "dataSfarsitTvaInc": "",
        "dataActualizareTvaInc": "",
        "dataPublicareTvaInc": "",
        "tipActTvaInc": "",
        "statusTvaIncasare": false
      },
      "stare_inactiv": {
        "dataInactivare": "",
        "dataReactivare": "",
        "dataPublicare": "",
        "dataRadiere": "",
        "statusInactivi": false
      },
      "inregistrare_SplitTVA": {
        "dataInceputSplitTVA": "",
        "dataAnulareSplitTVA": "",
        "statusSplitTVA": false
      },
      "adresa_sediu_social": {
        "sdenumire_Strada": "Str. Valea Cernatului",
        "snumar_Strada": "47",
        "sdenumire_Localitate": "Mun. Săcele",
        "scod_Localitate": "157",
        "sdenumire_Judet": "BRAŞOV",
        "scod_Judet": "8",
        "scod_JudetAuto": "BV",
        "stara": "",
        "sdetalii_Adresa": "",
        "scod_Postal": "505600"
      },
      "adresa_domiciliu_fiscal": {
        "ddenumire_Strada": "Str. Valea Cernatului",
        "dnumar_Strada": "47",
        "ddenumire_Localitate": "Mun. Săcele",
        "dcod_Localitate": "157",
        "ddenumire_Judet": "BRAŞOV",
        "dcod_Judet": "8",
        "dcod_JudetAuto": "BV",
        "dtara": "",
        "ddetalii_Adresa": "",
        "dcod_Postal": "505600"
      }
    }
  ],
  "notFound": []
}

More complex solution: https://github.com/agilegeeks/pyAnaf/blob/master/pyAnaf/api.py

from __future__ import unicode_literals, print_function

import datetime
import sys

PY_3_OR_HIGHER = sys.version_info >= (3, 0)

try:
    import urllib.request as urllib_request
    import urllib.error as urllib_error
except ImportError:
    import urllib2 as urllib_request
    import urllib2 as urllib_error

try:
    from cStringIO import StringIO
except ImportError:
    from io import BytesIO as StringIO

try:
    import http.client as http_client
except ImportError:
    import httplib as http_client

try:
    import json
except ImportError:
    import simplejson as json

from .models import AnafResultEntry


class AnafError(Exception):
    """
    Base Exception thrown by the Anaf object when there is a
    general error interacting with the API.
    """
    pass


class AnafHTTPError(Exception):
    """
    Exception thrown by the Anaf object when there is an
    HTTP error interacting with anaf.ro.
    """
    pass


class AnafResponseError(Exception):
    """
    Exception thrown by the Anaf object when there is an
    error the response returned from ANAF.
    """
    pass


class Anaf(object):
    WS_ENDPOINTS = {
        'sync': 'https://webservicesp.anaf.ro/PlatitorTvaRest/api/v3/ws/tva',
        'async': 'https://webservicesp.anaf.ro/AsynchWebService/api/v3/ws/tva'
    }
    LIMIT = 500

    def __init__(self):
        self.cuis = {}
        self.result = None
        self.entries = {}

    @staticmethod
    def _validate_cui(cui):
        if not isinstance(cui, int):
            raise AnafError('CUI should be integer')

    @staticmethod
    def _validate_date(date):
        if not isinstance(date, datetime.date):
            raise AnafError('Date should be of type datetime.date')

    @staticmethod
    def _prepare_data(data):
        if PY_3_OR_HIGHER:
            return bytes(data, 'utf-8')
        else:
            return data

    def addEndpoint(self, url, target='sync'):
        if target not in ['sync', 'async']:
            raise AnafError('Invalid target for endpoint. Must be one of \'sync\' or \'async\'')

        self.WS_ENDPOINTS[target] = url;

    def setLimit(self, limit):
        try:
            self.LIMIT = int(limit)
        except:
            raise AnafError('Limit should be an integer')

    def setCUIList(self, cui_list=[], date=None):
        if date is None:
            date = datetime.date.today()

        if len(cui_list) > self.LIMIT:
            raise AnafError('Too many CUIs to be queried. Should limit to %d' % self.LIMIT)

        self._validate_date(date)
        for cui in cui_list:
            self._validate_cui(cui)
            self.cuis[cui] = date

    def addCUI(self, cui, date=None):
        if date is None:
            date = datetime.date.today()

        self._validate_cui(cui)
        self._validate_date(date)

        self.cuis[cui] = date
        if len(self.cuis.items()) > self.LIMIT:
            raise AnafError('Too many CUIs to be queried. Should limit to %d' % self.LIMIT)

    def Request(self):
        # translate cuis entries to ANAF json format
        cui_list = []
        for entry in self.cuis.items():
            cui_list.append(
                {
                    'cui': entry[0],
                    'data': entry[1].isoformat()
                }
            )

        request = urllib_request.Request(self.WS_ENDPOINTS['sync'])
        request.add_header('Content-Type', 'application/json')

        try:
            response = urllib_request.urlopen(request, self._prepare_data(json.dumps(cui_list)))
        except urllib_error.HTTPError as e:
            raise AnafHTTPError('Error connecting to ANAF. Got a %d HTTP code.' % e.code)

        data = response.read()
        if isinstance(data, bytes):
            data = data.decode('utf-8')
        try:
            result = json.loads(data)
        except:
            raise AnafResponseError('Error parsing json response from ANAF.')

        if result['cod'] != 200:
            raise AnafResponseError('%s' % result['message'])

        result = result['found']
        self.result = result

        for entry in result:
            self.entries[entry['cui']] = AnafResultEntry(entry)

    def getCUIData(self, cui):
        if cui not in self.entries.keys():
            return None

        return self.entries[cui]

JavaScript Fetch PlatitorTvaRest Api

const url = 'https://webservicesp.anaf.ro/PlatitorTvaRest/api/v8/ws/tva';

const headers = {
    'Accept': 'application/json',
    'Accept-Language': 'en-US,en;q=0.7,ro;q=0.3',
    'Connection': 'keep-alive',
    'Referer': 'https://mfinante.gov.ro',
    'Content-Type': 'application/json',
};
const today = new Date().toISOString().slice(0, 10);
const data = [{'cui': '19467555', 'data': today}];
console.log(data)

fetch(url, {
    method: 'POST',
    headers: headers,
    body: JSON.stringify(data)
})
    .then(response => response.json())
    .then(data => console.log(data))
    .catch((error) => console.error('Error:', error));

*****
[{cui: '19467555', data: '2023-10-27'}]
{
"cod": 200,
"message": "SUCCESS",
"found": [
{
"date_generale": {
"cui": 19467555,
"data": "2023-10-27",
"denumire": "LEON TIBERIU CRISTIAN PERSOANĂ FIZICĂ AUTORIZATĂ",
"adresa": "JUD. BRAŞOV, MUN. SĂCELE, STR. VALEA CERNATULUI, NR.47",
"nrRegCom": "F08/388/2003",
"telefon": "0729947926",
"fax": "",
"codPostal": "505600",
"act": "",
"stare_inregistrare": "RELUARE ACTIVITATE din data 30.04.2013",
"data_inregistrare": "2007-01-01",
"cod_CAEN": "7430",
"iban": "",
"statusRO_e_Factura": false,
"organFiscalCompetent": "Administraţia Judeţeană a Finanţelor Publice Braşov",
"forma_de_proprietate": "",
"forma_organizare": "",
"forma_juridica": ""
},
"inregistrare_scop_Tva": {
"scpTVA": false,
"perioade_TVA": []
},
"inregistrare_RTVAI": {
"dataInceputTvaInc": "",
"dataSfarsitTvaInc": "",
"dataActualizareTvaInc": "",
"dataPublicareTvaInc": "",
"tipActTvaInc": "",
"statusTvaIncasare": false
},
"stare_inactiv": {
"dataInactivare": "",
"dataReactivare": "",
"dataPublicare": "",
"dataRadiere": "",
"statusInactivi": false
},
"inregistrare_SplitTVA": {
"dataInceputSplitTVA": "",
"dataAnulareSplitTVA": "",
"statusSplitTVA": false
},
"adresa_sediu_social": {
"sdenumire_Strada": "Str. Valea Cernatului",
"snumar_Strada": "47",
"sdenumire_Localitate": "Mun. Săcele",
"scod_Localitate": "157",
"sdenumire_Judet": "BRAŞOV",
"scod_Judet": "8",
"scod_JudetAuto": "BV",
"stara": "",
"sdetalii_Adresa": "",
"scod_Postal": "505600"
},
"adresa_domiciliu_fiscal": {
"ddenumire_Strada": "Str. Valea Cernatului",
"dnumar_Strada": "47",
"ddenumire_Localitate": "Mun. Săcele",
"dcod_Localitate": "157",
"ddenumire_Judet": "BRAŞOV",
"dcod_Judet": "8",
"dcod_JudetAuto": "BV",
"dtara": "",
"ddetalii_Adresa": "",
"dcod_Postal": "505600"
}
}
],
"notFound": []
}