Seulement le minimum nécessaire pour les utiliser avec **FastAPI** sera couvert... ce qui est en réalité très peu.

**FastAPI** est totalement basé sur ces annotations de type, qui lui donnent de nombreux avantages.

Mais même si vous n'utilisez pas ou n'utiliserez jamais **FastAPI**, vous pourriez bénéficier d'apprendre quelques choses sur ces dernières.

!!! note

)

// LookupUserDN searches for the full DN and groups of a given short/login
// username.
func (l *Config) LookupUserDN(username string) (string, []string, error) {
	conn, err := l.LDAP.Connect()
	if err != nil {
		return "", nil, err
	}
	defer conn.Close()

	// Bind to the lookup user account
	if err = l.LDAP.LookupBind(conn); err != nil {
		return "", nil, err
	}

	// Lookup user DN

	ztunnelKeepAliveCheckInterval = time.Second / 10
	mt := monitortest.New(t)
	setupLogging()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	fixture := connect(ctx)
	ztunClient := fixture.ztunClient
	uid := fixture.uid

	m, fds := readRequest(t, ztunClient)
	// we got am essage from ztun, so it should have observed us being connected

                    Hexdump.hexdump( log, sbuf, 4, n );
                }
            }
        }
    }
    public void connect() throws SmbException {
        try {
            super.connect( RESPONSE_TIMEOUT );
        } catch( TransportException te ) {
            throw new SmbException( "Failed to connect: " + address, te );
        }
    }
    protected void doConnect() throws IOException {
        /*

        super(connection.getURL());
        this.connection = connection;
        requestProperties = new HashMap();
    }

    public void connect() throws IOException {
        if (connected) return;
        connection.connect();
        connected = true;
    }

    private void handshake() throws IOException {
        if (handshakeComplete) return;
        doHandshake();

### [Addresses](https://square.github.io/okhttp/4.x/okhttp/okhttp3/-address/)

Addresses specify a webserver (like `github.com`) and all of the **static** configuration necessary to connect to that server: the port number, HTTPS settings, and preferred network protocols (like HTTP/2).

    return new TunnelingUnixSocket(path, channel);
  }

  @Override public Socket createSocket(String host, int port) throws IOException {
    Socket result = createSocket();

    try {
      result.connect(new InetSocketAddress(host, port));
    } catch (IOException e) {
      result.close();
      throw e;
    }
    return result;
  }

  @Override public Socket createSocket(

engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")

      "version_info": "2024-03-04T08:37:44Z/4",
      "dynamic_listeners": [
        {
          "name": "connect_terminate",
          "active_state": {
            "listener": {
              "@type": "type.googleapis.com/envoy.config.listener.v3.Listener",
              "name": "connect_terminate",
              "address": {
                "socket_address": {
                  "address": "0.0.0.0",

def test_websocket():
    with client.websocket_connect("/items/portal-gun") as websocket:
        data = websocket.receive_json()
        assert data == {"item_id": "portal-gun", "path": "/items/{item_id}"}


def test_websocket_invalid_path_doesnt_match():
    with pytest.raises(WebSocketDisconnect):
        with client.websocket_connect("/itemsx/portal-gun"):