}
	endpointLocal, err := NewEndpoint("http://localhost:9012")
	if err != nil {
		t.Fatalf("unexpected error %v", err)
	}
	endpointLocal.IsLocal = true

	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()
	err = initGlobalLockGrid(ctx, []PoolEndpoints{{Endpoints: Endpoints{endpoint, endpointLocal}}})
	if err != nil {
		t.Fatal(err)
	}

	lkClient := newlockRESTClient(endpoint)
	if lkClient.IsOnline() {

# WebSockets { #websockets }

Vous pouvez utiliser [WebSockets](https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API) avec **FastAPI**.

## Installer `websockets` { #install-websockets }

Vous devez créer un [environnement virtuel](../virtual-environments.md), l'activer, et installer `websockets` (une bibliothèque Python qui facilite l'utilisation du protocole « WebSocket ») :

<div class="termy">

```console
$ pip install websockets

	if err != nil {
		return err
	}
	m[policy] = p
	return nil
}

func (iamOS *IAMObjectStore) loadPolicyDocs(ctx context.Context, m map[string]PolicyDoc) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
	for item := range listIAMConfigItems(ctx, iamOS.objAPI, iamConfigPoliciesPrefix) {
		if item.Err != nil {
			return item.Err
		}

		policyName := path.Dir(item.Item)

	qs.TgtXferStats = make(map[string]map[RMetricName]XferStats)
	qs.MRFStats = ReplicationMRFStats{
		LastFailedCount: atomic.LoadUint64(&r.mrfStats.LastFailedCount),
	}

	r.RLock()
	defer r.RUnlock()

	brs, ok := r.Cache[bucket]
	if !ok {
		return qs
	}
	for arn := range brs.Stats {
		qs.TgtXferStats[arn] = make(map[RMetricName]XferStats)
	}
	count := 0
	var totPeak float64

func WithDeadline[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error)) (result V, err error) {
	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	c := make(chan ioret[V], 1)
	go func() {
		v, err := work(ctx)
		c <- ioret[V]{val: v, err: err}
	}()

	select {
	case v := <-c:
		return v.val, v.err
	case <-ctx.Done():

```Python
UserInDB(
    username = user_dict["username"],
    password = user_dict["password"],
    email = user_dict["email"],
    full_name = user_dict["full_name"],
)
```

#### Créer un modèle Pydantic à partir du contenu d'un autre { #a-pydantic-model-from-the-contents-of-another }

Comme dans l'exemple ci-dessus nous avons obtenu `user_dict` depuis `user_in.model_dump()`, ce code :

```Python

		cycleInfo.next = binary.LittleEndian.Uint64(buf[:8])
		buf = buf[8:]
		_, err := cycleInfo.UnmarshalMsg(buf)
		bugLogIf(ctx, err)
	}

	scannerTimer := time.NewTimer(scannerCycle.Load())
	defer scannerTimer.Stop()
	defer globalScannerMetrics.setCycle(nil)

	for {
		select {
		case <-ctx.Done():
			return
		case <-scannerTimer.C:
			// Reset the timer for next cycle.

/// info

Pour recevoir des fichiers téléversés et/ou des données de formulaire, installez d'abord [`python-multipart`](https://github.com/Kludex/python-multipart).

Vous devez créer un [environnement virtuel](../virtual-environments.md), l'activer, puis installer ce paquet, par exemple :

```console
$ pip install python-multipart
```

///

## Importer `File` et `Form` { #import-file-and-form }

type bootstrapTracer struct {
	mu   sync.RWMutex
	info []madmin.TraceInfo
}

var globalBootstrapTracer = &bootstrapTracer{}

func (bs *bootstrapTracer) Record(info madmin.TraceInfo) {
	bs.mu.Lock()
	defer bs.mu.Unlock()

	if len(bs.info) > bootstrapTraceLimit {
		return
	}
	bs.info = append(bs.info, info)
}

func (bs *bootstrapTracer) Events() []madmin.TraceInfo {

func CheckPortAvailability(host, port string, opts TCPOptions) (err error) {
	lc := &net.ListenConfig{}

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	l, err := lc.Listen(ctx, "tcp", net.JoinHostPort(host, port))
	if err != nil {
		return err
	}

	// As we are able to listen on this network, the port is not in use.